Merge pull request #71 from camunda-community-hub/fm-exponential-backoff

Add backoff to retrying
camunda-community-hub · Jan 26, 2022 · d601d66 · d601d66
2 parents 793c126 + 02a8a1e
commit d601d66
Show file tree

Hide file tree

Showing 4 changed files with 30 additions and 6 deletions.
diff --git a/pom.xml b/pom.xml
@@ -14,7 +14,7 @@
   <parent>
       <groupId>org.camunda.community</groupId>
       <artifactId>community-hub-release-parent</artifactId>
-      <version>1.2.2</version>    
+      <version>1.2.2</version>
       <relativePath />
   </parent>
 
@@ -40,7 +40,7 @@
     <version.junit>5.5.2</version.junit>
     <version.log4j>2.17.1</version.log4j>
     <version.slf4j>1.7.26</version.slf4j>
-    <version.zeebe>1.2.1</version.zeebe>
+    <version.zeebe>1.3.1</version.zeebe>
 
     <!-- plugin version -->
     <plugin.version.assembly>3.1.1</plugin.version.assembly>

diff --git a/src/main/java/io/zeebe/kafka/connect/sink/ZeebeSinkFuture.java b/src/main/java/io/zeebe/kafka/connect/sink/ZeebeSinkFuture.java
@@ -17,11 +17,15 @@
 
 import io.camunda.zeebe.client.api.command.FinalCommandStep;
 import io.camunda.zeebe.client.api.response.PublishMessageResponse;
+import io.camunda.zeebe.client.api.worker.BackoffSupplier;
 import io.grpc.Status.Code;
 import io.grpc.StatusRuntimeException;
 import java.util.EnumSet;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -51,8 +55,15 @@ class ZeebeSinkFuture extends CompletableFuture<PublishMessageResponse> {
 
   private final FinalCommandStep<PublishMessageResponse> command;
 
-  ZeebeSinkFuture(final FinalCommandStep<PublishMessageResponse> command) {
+  // TODO inject reusable executor
+  private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+  private BackoffSupplier backoffSupplier;
+  private long currentRetryDelay = 50L;
+
+  ZeebeSinkFuture(
+      final FinalCommandStep<PublishMessageResponse> command, BackoffSupplier backoffSupplier) {
     this.command = command;
+    this.backoffSupplier = backoffSupplier;
   }
 
   @SuppressWarnings("unchecked")
@@ -73,7 +84,9 @@ CompletableFuture<PublishMessageResponse> executeAsync() {
                 if (SUCCESS_CODES.contains(code)) {
                   complete(null);
                 } else if (RETRIABLE_CODES.contains(code)) {
-                  executeAsync();
+                  currentRetryDelay = backoffSupplier.supplyRetryDelay(currentRetryDelay);
+                  LOGGER.trace("Retry " + command + " after seeing " + code + ", backoff/delay: " + currentRetryDelay + " ms");
+                  executor.schedule(this::executeAsync, currentRetryDelay, TimeUnit.MILLISECONDS);
                 } else if (FAILURE_CODES.contains(code)) {
                   completeExceptionally(throwable);
                 } else {

diff --git a/src/main/java/io/zeebe/kafka/connect/sink/ZeebeSinkTask.java b/src/main/java/io/zeebe/kafka/connect/sink/ZeebeSinkTask.java
@@ -19,6 +19,8 @@
 import io.camunda.zeebe.client.api.command.FinalCommandStep;
 import io.camunda.zeebe.client.api.command.PublishMessageCommandStep1.PublishMessageCommandStep3;
 import io.camunda.zeebe.client.api.response.PublishMessageResponse;
+import io.camunda.zeebe.client.api.worker.BackoffSupplier;
+import io.camunda.zeebe.client.impl.worker.ExponentialBackoffBuilderImpl;
 import io.zeebe.kafka.connect.sink.message.JsonRecordParser;
 import io.zeebe.kafka.connect.sink.message.JsonRecordParser.Builder;
 import io.zeebe.kafka.connect.sink.message.Message;
@@ -41,12 +43,20 @@ public class ZeebeSinkTask extends SinkTask {
 
   private ManagedClient managedClient;
   private JsonRecordParser parser;
+  private BackoffSupplier backoffSupplier;
 
   @Override
   public void start(final Map<String, String> props) {
     final ZeebeSinkConnectorConfig config = new ZeebeSinkConnectorConfig(props);
     managedClient = new ManagedClient(ZeebeClientHelper.buildClient(config));
     parser = buildParser(config);
+    backoffSupplier =
+        new ExponentialBackoffBuilderImpl()
+            .maxDelay(1000L)
+            .minDelay(50L)
+            .backoffFactor(1.5)
+            .jitterFactor(0.2)
+            .build();
   }
 
   // The documentation specifies that we probably shouldn't block here but I'm not sure what the
@@ -82,7 +92,7 @@ private CompletableFuture<Void> publishMessages(
         sinkRecords
             .stream()
             .map(r -> this.preparePublishRequest(client, r))
-            .map(ZeebeSinkFuture::new)
+            .map(command -> new ZeebeSinkFuture(command, backoffSupplier))
             .map(ZeebeSinkFuture::executeAsync)
             .toArray(CompletableFuture[]::new);
 

diff --git a/src/test/java/io/zeebe/kafka/connect/sink/ZeebeSinkFutureTest.java b/src/test/java/io/zeebe/kafka/connect/sink/ZeebeSinkFutureTest.java
@@ -21,6 +21,7 @@
 import io.camunda.zeebe.client.api.ZeebeFuture;
 import io.camunda.zeebe.client.api.command.FinalCommandStep;
 import io.camunda.zeebe.client.api.response.PublishMessageResponse;
+import io.camunda.zeebe.client.impl.worker.ExponentialBackoffBuilderImpl;
 import io.grpc.Status;
 import io.grpc.Status.Code;
 import io.grpc.StatusRuntimeException;
@@ -130,7 +131,7 @@ private ZeebeSinkFuture create(final RuntimeException e) {
   }
 
   private ZeebeSinkFuture create(final Runnable r) {
-    return new ZeebeSinkFuture(new FinalStepMock(r));
+    return new ZeebeSinkFuture(new FinalStepMock(r), new ExponentialBackoffBuilderImpl().build());
   }
 
   static class FinalStepMock implements FinalCommandStep<PublishMessageResponse> {