(feat) add retry on failed sync commit

Author: ylgrgyq

src/main/java/cn/leancloud/kafka/consumer/AbstractCommitPolicy.java

@@ -4,7 +4,9 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.RetriableException;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -15,14 +17,50 @@
 import static java.util.stream.Collectors.toSet;
 
 abstract class AbstractCommitPolicy<K, V> implements CommitPolicy<K, V> {
-    protected final Consumer<K, V> consumer;
+    static SleepFunction sleepFunction = Thread::sleep;
+
+    interface SleepFunction {
+        void sleep(long timeout) throws InterruptedException;
+    }
+
+    private static class RetryContext {
+        private final long retryInterval;
+        private final int maxAttempts;
+        private int numOfAttempts;
+
+        private RetryContext(long retryInterval, int maxAttempts) {
+            this.retryInterval = retryInterval;
+            this.maxAttempts = maxAttempts;
+            this.numOfAttempts = 0;
+        }
+
+        void onError(RetriableException e) {
+            if (++numOfAttempts >= maxAttempts) {
+                throw e;
+            } else {
+                try {
+                    sleepFunction.sleep(retryInterval);
+                } catch (InterruptedException ex) {
+                    e.addSuppressed(ex);
+                    Thread.currentThread().interrupt();
+                    throw e;
+                }
+            }
+        }
+    }
+
     final Map<TopicPartition, Long> topicOffsetHighWaterMark;
     final Map<TopicPartition, OffsetAndMetadata> completedTopicOffsets;
+    protected final Consumer<K, V> consumer;
+    private final long syncCommitRetryIntervalMs;
+    private final int maxAttemptsForEachSyncCommit;
 
-    AbstractCommitPolicy(Consumer<K, V> consumer) {
+    AbstractCommitPolicy(Consumer<K, V> consumer, Duration syncCommitRetryInterval, int maxAttemptsForEachSyncCommit) {
         this.consumer = consumer;
         this.topicOffsetHighWaterMark = new HashMap<>();
         this.completedTopicOffsets = new HashMap<>();
+        this.syncCommitRetryIntervalMs = syncCommitRetryInterval.toMillis();
+        this.maxAttemptsForEachSyncCommit = maxAttemptsForEachSyncCommit;
     }
 
     @Override
@@ -43,7 +81,7 @@ public void markCompletedRecord(ConsumerRecord<K, V> record) {
 
     @Override
     public Set<TopicPartition> syncPartialCommit() {
-        consumer.commitSync(completedTopicOffsets);
+        commitSync(completedTopicOffsets);
         final Set<TopicPartition> partitions = checkCompletedPartitions();
         completedTopicOffsets.clear();
         for (TopicPartition p : partitions) {
@@ -85,6 +123,30 @@ Map<TopicPartition, OffsetAndMetadata> completedTopicOffsets() {
         return completedTopicOffsets;
     }
 
+    void commitSync() {
+        final RetryContext context = context();
+        do {
+            try {
+                consumer.commitSync();
+                return;
+            } catch (RetriableException e) {
+                context.onError(e);
+            }
+        } while (true);
+    }
+
+    void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets) {
+        final RetryContext context = context();
+        do {
+            try {
+                consumer.commitSync(offsets);
+                return;
+            } catch (RetriableException e) {
+                context.onError(e);
+            }
+        } while (true);
+    }
+
     private Set<TopicPartition> checkCompletedPartitions() {
         return completedTopicOffsets
                 .entrySet()
@@ -102,4 +164,8 @@ private boolean topicOffsetMeetHighWaterMark(TopicPartition topicPartition, Offs
         // maybe this partition revoked before a msg of this partition was processed
         return true;
     }
+
+    private RetryContext context() {
+        return new RetryContext(syncCommitRetryIntervalMs, maxAttemptsForEachSyncCommit);
+    }
 }

src/main/java/cn/leancloud/kafka/consumer/AbstractRecommitAwareCommitPolicy.java

@@ -12,8 +12,11 @@ abstract class AbstractRecommitAwareCommitPolicy<K, V> extends AbstractCommitPol
     private final Duration recommitInterval;
     private long nextRecommitNanos;
 
-    AbstractRecommitAwareCommitPolicy(Consumer<K, V> consumer, Duration recommitInterval) {
-        super(consumer);
+    AbstractRecommitAwareCommitPolicy(Consumer<K, V> consumer,
+                                      Duration syncCommitRetryInterval,
+                                      int maxAttemptsForEachSyncCommit,
+                                      Duration recommitInterval) {
+        super(consumer, syncCommitRetryInterval, maxAttemptsForEachSyncCommit);
         this.recommitInterval = recommitInterval;
         updateNextRecommitTime(System.nanoTime());
     }

src/main/java/cn/leancloud/kafka/consumer/AsyncCommitPolicy.java

@@ -21,8 +21,12 @@ final class AsyncCommitPolicy<K, V> extends AbstractRecommitAwareCommitPolicy<K,
     private int pendingAsyncCommitCounter;
     private boolean forceSync;
 
-    AsyncCommitPolicy(Consumer<K, V> consumer, Duration recommitInterval, int maxPendingAsyncCommits) {
-        super(consumer, recommitInterval);
+    AsyncCommitPolicy(Consumer<K, V> consumer,
+                      Duration syncCommitRetryInterval,
+                      int maxAttemptsForEachSyncCommit,
+                      Duration recommitInterval,
+                      int maxPendingAsyncCommits) {
+        super(consumer, syncCommitRetryInterval, maxAttemptsForEachSyncCommit, recommitInterval);
         this.maxPendingAsyncCommits = maxPendingAsyncCommits;
         this.callback = new AsyncCommitCallback();
     }
@@ -91,9 +95,9 @@ private void asyncCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
 
     private void syncCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
         if (offsets.isEmpty()) {
-            consumer.commitSync();
+            commitSync();
         } else {
-            consumer.commitSync(offsets);
+            commitSync(offsets);
         }
     }
 

src/main/java/cn/leancloud/kafka/consumer/AutoCommitPolicy.java

@@ -9,7 +9,8 @@
 
 final class AutoCommitPolicy<K, V> extends AbstractCommitPolicy<K, V> {
     AutoCommitPolicy(Consumer<K, V> consumer) {
-        super(consumer);
+        // For auto commit policy we don't commit by hand so we don't need retry sync commit facilities
+        super(consumer, Duration.ZERO, 1);
     }
 
     @Override

src/main/java/cn/leancloud/kafka/consumer/LcKafkaConsumerBuilder.java

@@ -1,9 +1,6 @@
 package cn.leancloud.kafka.consumer;
 
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.consumer.*;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -85,6 +82,8 @@ private static void requireArgument(boolean expression, String template, Object.
     private Map<String, Object> configs;
     private ConsumerRecordHandler<K, V> consumerRecordHandler;
     private ConsumerSeekDestination forceSeekTo = ConsumerSeekDestination.NONE;
+    private Duration syncCommitRetryInterval = Duration.ofSeconds(1);
+    private int maxAttemptsForEachSyncCommit = 3;
     @Nullable
     private Consumer<K, V> consumer;
     @Nullable
@@ -235,6 +234,83 @@ public LcKafkaConsumerBuilder<K, V> gracefulShutdownTimeout(Duration gracefulShu
         return this;
     }
 
+    /**
+     * Sets the amount of time to wait before retry a failed synchronous commit on calling {@link KafkaConsumer#commitSync()}.
+     * or {@link KafkaConsumer#commitSync(Map)}. Every synchronous commit may fail but most of times they are caused by
+     * {@link org.apache.kafka.common.errors.RetriableException} and we can retry commit on this kind of exception safely.
+     * This configuration set the interval between each retry.
+     * <p>
+     * For those failures of asynchronous commit by calling {@link KafkaConsumer#commitAsync()} or
+     * {@link KafkaConsumer#commitAsync(OffsetCommitCallback)}, we retry them by a synchronous commit automatically
+     * when we found any of them. So we only need configurations for synchronous commits.
+     * <p>
+     * The default {@code syncCommitRetryIntervalMillis} is 1000.
+     *
+     * @param syncCommitRetryIntervalMillis the retry interval in milliseconds
+     * @return this
+     * @throws IllegalArgumentException if {@code syncCommitRetryIntervalMillis} is a negative value
+     */
+    public LcKafkaConsumerBuilder<K, V> syncCommitRetryIntervalMillis(long syncCommitRetryIntervalMillis) {
+        requireArgument(syncCommitRetryIntervalMillis >= 0,
+                "syncCommitRetryIntervalMillis: %s (expected >= 0)", syncCommitRetryIntervalMillis);
+        this.syncCommitRetryInterval = Duration.ofMillis(syncCommitRetryIntervalMillis);
+        return this;
+    }
+
+    /**
+     * Sets the amount of time to wait before retry a failed synchronous commit on calling {@link KafkaConsumer#commitSync()}.
+     * or {@link KafkaConsumer#commitSync(Map)}. Every synchronous commit may fail but most of times they are caused by
+     * {@link org.apache.kafka.common.errors.RetriableException} and we can retry commit on this kind of exception safely.
+     * This configuration set the interval between each retry.
+     * <p>
+     * For those failures of asynchronous commit by calling {@link KafkaConsumer#commitAsync()} or
+     * {@link KafkaConsumer#commitAsync(OffsetCommitCallback)}, we retry them by a synchronous commit automatically
+     * when we found any of them. So we only need configurations for synchronous commits.
+     * <p>
+     * The default {@code syncCommitRetryInterval} is 1 second.
+     *
+     * @param syncCommitRetryInterval the retry interval
+     * @return this
+     * @throws NullPointerException     if {@code syncCommitRetryInterval} is null
+     * @throws IllegalArgumentException if {@code syncCommitRetryInterval} is a negative duration
+     */
+    public LcKafkaConsumerBuilder<K, V> syncCommitRetryInterval(Duration syncCommitRetryInterval) {
+        requireNonNull(syncCommitRetryInterval, "syncCommitRetryInterval");
+        requireArgument(!syncCommitRetryInterval.isNegative(),
+                "syncCommitRetryInterval: %s (expect positive or zero duration)", syncCommitRetryInterval);
+        this.syncCommitRetryInterval = syncCommitRetryInterval;
+        return this;
+    }
+
+    /**
+     * Sets the maximum attempt times for a synchronous commit by calling {@link KafkaConsumer#commitSync()}.
+     * or {@link KafkaConsumer#commitSync(Map)}. Every synchronous commit may fail but most of times they are caused by
+     * {@link org.apache.kafka.common.errors.RetriableException} and we can retry commit on this kind of exception safely.
+     * This configuration cap the maximum retry times. If attempts reach to {@code maxAttemptsForEachSyncCommit}, the cached
+     * {@link org.apache.kafka.common.errors.RetriableException} will be rethrown by then it will cause the Kafka Consumer
+     * to stop and quit.
+     * <p>
+     * For those failures of asynchronous commit by calling {@link KafkaConsumer#commitAsync()} or
+     * {@link KafkaConsumer#commitAsync(OffsetCommitCallback)}, we retry them by a synchronous commit automatically
+     * when we found any of them. So we only need configurations for synchronous commits.
+     * <p>
+     * Please note that {@code maxAttemptsForEachSyncCommit} multiplies
+     * {@code syncCommitRetryInterval} should far lower than {@code max.poll.interval.ms}, otherwise Kafka Consumer
+     * may encounter session timeout or polling timeout due to not calling {@link KafkaConsumer#poll(long)} for too long.
+     * <p>
+     * The default {@code maxAttemptsForEachSyncCommit} is 3.
+     *
+     * @param maxAttemptsForEachSyncCommit maximum attempt times for a synchronous commit
+     * @return this
+     * @throws IllegalArgumentException if {@code maxAttemptsForEachSyncCommit} is zero or a negative value
+     */
+    public LcKafkaConsumerBuilder<K, V> maxAttemptsForEachSyncCommit(int maxAttemptsForEachSyncCommit) {
+        requireArgument(maxAttemptsForEachSyncCommit > 0,
+                "maxAttemptsForEachSyncCommit: %s (expect > 0)", maxAttemptsForEachSyncCommit);
+        this.maxAttemptsForEachSyncCommit = maxAttemptsForEachSyncCommit;
+        return this;
+    }
+
     /**
      * When using async consumer to commit offset asynchronously, this argument can force consumer to do a synchronous
      * commit after there's already this ({@code maxPendingAsyncCommits}) many async commits on the fly without
@@ -407,7 +483,8 @@ public LcKafkaConsumerBuilder<K, V> forceSeekTo(ConsumerSeekDestination destinat
     public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildAuto() {
         checkConfigs(AutoCommitConsumerConfigs.values());
         consumer = buildConsumer(true);
-        policy = workerPool == ImmediateExecutorService.INSTANCE ? NoOpCommitPolicy.getInstance() : new AutoCommitPolicy<>(consumer);
+        policy = workerPool == ImmediateExecutorService.INSTANCE ?
+                NoOpCommitPolicy.getInstance() : new AutoCommitPolicy<>(consumer);
         return doBuild();
     }
 
@@ -435,7 +512,11 @@ public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildAuto() {
      */
     public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildSync() {
         consumer = buildConsumer(false);
-        policy = new SyncCommitPolicy<>(consumer, getRecommitInterval());
+        policy = new SyncCommitPolicy<>(
+                consumer,
+                syncCommitRetryInterval,
+                maxAttemptsForEachSyncCommit,
+                getRecommitInterval());
         return doBuild();
     }
 
@@ -462,7 +543,11 @@ public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildSync() {
      */
     public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildPartialSync() {
         consumer = buildConsumer(false);
-        policy = new PartialSyncCommitPolicy<>(consumer, getRecommitInterval());
+        policy = new PartialSyncCommitPolicy<>(
+                consumer,
+                syncCommitRetryInterval,
+                maxAttemptsForEachSyncCommit,
+                getRecommitInterval());
         return doBuild();
     }
 
@@ -494,7 +579,12 @@ public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildPartialSync() {
      */
     public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildAsync() {
         consumer = buildConsumer(false);
-        policy = new AsyncCommitPolicy<>(consumer, getRecommitInterval(), maxPendingAsyncCommits);
+        policy = new AsyncCommitPolicy<>(
+                consumer,
+                syncCommitRetryInterval,
+                maxAttemptsForEachSyncCommit,
+                getRecommitInterval(),
+                maxPendingAsyncCommits);
         return doBuild();
     }
 
@@ -525,7 +615,12 @@ public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildAsync() {
      */
     public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildPartialAsync() {
         consumer = buildConsumer(false);
-        policy = new PartialAsyncCommitPolicy<>(consumer, getRecommitInterval(), maxPendingAsyncCommits);
+        policy = new PartialAsyncCommitPolicy<>(
+                consumer,
+                syncCommitRetryInterval,
+                maxAttemptsForEachSyncCommit,
+                getRecommitInterval(),
+                maxPendingAsyncCommits);
         return doBuild();
     }
 

src/main/java/cn/leancloud/kafka/consumer/PartialAsyncCommitPolicy.java

@@ -19,8 +19,12 @@ final class PartialAsyncCommitPolicy<K, V> extends AbstractRecommitAwareCommitPo
     private int pendingAsyncCommitCounter;
     private boolean forceSync;
 
-    PartialAsyncCommitPolicy(Consumer<K, V> consumer, Duration forceWholeCommitInterval, int maxPendingAsyncCommits) {
-        super(consumer, forceWholeCommitInterval);
+    PartialAsyncCommitPolicy(Consumer<K, V> consumer,
+                             Duration syncCommitRetryInterval,
+                             int maxAttemptsForEachSyncCommit,
+                             Duration forceWholeCommitInterval,
+                             int maxPendingAsyncCommits) {
+        super(consumer, syncCommitRetryInterval, maxAttemptsForEachSyncCommit, forceWholeCommitInterval);
         this.maxPendingAsyncCommits = maxPendingAsyncCommits;
         this.callback = new AsyncCommitCallback();
         this.pendingAsyncCommitOffset = new HashMap<>();
@@ -36,7 +40,7 @@ public Set<TopicPartition> tryCommit(boolean noPendingRecords) {
         } else {
             final Set<TopicPartition> partitions = getCompletedPartitions(noPendingRecords);
             if (syncCommit) {
-                consumer.commitSync(offsets);
+                commitSync(offsets);
                 pendingAsyncCommitOffset.clear();
                 pendingAsyncCommitCounter = 0;
                 forceSync = false;

src/main/java/cn/leancloud/kafka/consumer/PartialSyncCommitPolicy.java

@@ -10,15 +10,18 @@
 import java.util.Set;
 
 final class PartialSyncCommitPolicy<K, V> extends AbstractRecommitAwareCommitPolicy<K, V> {
-    PartialSyncCommitPolicy(Consumer<K, V> consumer, Duration forceWholeCommitInterval) {
-        super(consumer, forceWholeCommitInterval);
+    PartialSyncCommitPolicy(Consumer<K, V> consumer,
+                            Duration syncCommitRetryInterval,
+                            int maxAttemptsForEachSyncCommit,
+                            Duration forceWholeCommitInterval) {
+        super(consumer, syncCommitRetryInterval, maxAttemptsForEachSyncCommit, forceWholeCommitInterval);
     }
 
     @Override
     public Set<TopicPartition> tryCommit(boolean noPendingRecords) {
         final Map<TopicPartition, OffsetAndMetadata> offsets = offsetsForPartialCommit();
         if (!offsets.isEmpty()) {
-            consumer.commitSync(offsets);
+            commitSync(offsets);
         }
 
         if (completedTopicOffsets.isEmpty()) {