a simple back pressure mechanism

Author: Mike Skells

gcs-sink-connector/src/main/java/io/aiven/kafka/connect/gcs/GcsSinkTask.java

@@ -49,6 +49,8 @@ public final class GcsSinkTask extends SinkTask {
     private GcsSinkConfig config;
 
     private Storage storage;
+    private int backPressureHardLimit;
+    private int backPressureCurrentBuffer;
 
     // required by Connect
     public GcsSinkTask() {
@@ -88,6 +90,7 @@ public void start(final Map<String, String> props) {
         if (Objects.nonNull(config.getKafkaRetryBackoffMs())) {
             context.timeout(config.getKafkaRetryBackoffMs());
         }
+        backPressureHardLimit = 100_000;// TODO : this shoul dbe from some config - config.getBackPressureHardLimit();
     }
 
     private void initRest() {
@@ -106,6 +109,11 @@ public void put(final Collection<SinkRecord> records) {
         for (final SinkRecord record : records) {
             recordGrouper.put(record);
         }
+        backPressureCurrentBuffer += records.size();
+        if (backPressureCurrentBuffer >= backPressureHardLimit) {
+            LOG.warn("Back pressure limit reached, requesting flush");
+            context.requestCommit();
+        }
     }
 
     @Override
@@ -115,6 +123,7 @@ public void flush(final Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
         } finally {
             recordGrouper.clear();
         }
+        backPressureCurrentBuffer = 0;
     }
 
     private void flushFile(final String filename, final List<SinkRecord> records) {