START: DRAFT: Custom retry delay provider

Author: astubbs

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelConsumerOptions.java

@@ -12,6 +12,7 @@
 
 import java.time.Duration;
 import java.util.Objects;
+import java.util.function.Function;
 
 import static io.confluent.csid.utils.StringUtils.msg;
 import static io.confluent.parallelconsumer.ParallelConsumerOptions.CommitMode.PERIODIC_TRANSACTIONAL_PRODUCER;
@@ -131,6 +132,16 @@ public enum CommitMode {
     @Builder.Default
     private final Duration defaultMessageRetryDelay = Duration.ofSeconds(1);
 
+    /**
+     * When present, use this to generate the retry delay, instad of {@link #getDefaultMessageRetryDelay()}.
+     * <p>
+     * Overrides {@link #defaultMessageRetryDelay}, even if it's set.
+     */
+    @Builder.Default
+    private final Function<WorkContainer, Duration> retryDelayProvider;
+
+    static Function<WorkContainer, Duration> retryDelayProviderStatic;
+
     public void validate() {
         Objects.requireNonNull(consumer, "A consumer must be supplied");
 
@@ -141,6 +152,7 @@ public void validate() {
 
         //
         WorkContainer.setDefaultRetryDelay(getDefaultMessageRetryDelay());
+        ParallelConsumerOptions.retryDelayProviderStatic = getRetryDelayProvider();
     }
 
     protected boolean isUsingTransactionalProducer() {

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelEoSStreamProcessor.java

@@ -32,7 +32,6 @@
 import static io.confluent.csid.utils.StringUtils.msg;
 import static io.confluent.parallelconsumer.UserFunctions.carefullyRun;
 import static java.time.Duration.ofMillis;
-import static java.time.Duration.ofSeconds;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.SECONDS;
 

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/WorkContainer.java

@@ -18,6 +18,7 @@
 import java.util.Optional;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 
 import static io.confluent.csid.utils.KafkaUtils.toTP;
 
@@ -59,6 +60,9 @@ public class WorkContainer<K, V> implements Comparable<WorkContainer> {
      */
     private Duration retryDelay;
 
+    /**
+     * @see ParallelConsumerOptions#getDefaultMessageRetryDelay()
+     */
     @Setter
     static Duration defaultRetryDelay = Duration.ofSeconds(1);
 
@@ -121,10 +125,15 @@ private Temporal tryAgainAt(WallClock clock) {
     }
 
     public Duration getRetryDelay() {
-        if (retryDelay == null)
-            return defaultRetryDelay;
-        else
-            return retryDelay;
+        var retryDelayProvider = ParallelConsumerOptions.retryDelayProviderStatic;
+        if (retryDelayProvider != null) {
+            return retryDelayProvider.apply(this);
+        } else {
+            if (retryDelay == null)
+                return defaultRetryDelay;
+            else
+                return retryDelay;
+        }
     }
 
     @Override