Fix consumer timeout after broker restart

wushilin · Jul 25, 2024 · 68c0d0c · 68c0d0c
1 parent ab277c9
commit 68c0d0c
Showing 1 changed file with 14 additions and 1 deletion.
diff --git a/...er-core/src/main/java/io/confluent/parallelconsumer/internal/ConsumerOffsetCommitter.java b/...er-core/src/main/java/io/confluent/parallelconsumer/internal/ConsumerOffsetCommitter.java
@@ -90,7 +90,20 @@ protected void commitOffsets(final Map<TopicPartition, OffsetAndMetadata> offset
         switch (commitMode) {
             case PERIODIC_CONSUMER_SYNC -> {
                 log.debug("Committing offsets Sync");
-                consumerMgr.commitSync(offsetsToSend);
+                while(true) {
+                    try {
+                        consumerMgr.commitSync(offsetsToSend);
+                        //break when offset commit is okay, never throw exception to the main threads
+                        break;
+                    } catch(Throwable t) {
+                        log.error("Failed to commit offset. Retrying in 10 seconds", t);
+                        try {
+                            Thread.sleep(10000L);
+                        } catch(InterruptedException ite) {
+                            log.info("Giving up offset commit due to interruption");
+                        }
+                    }
+                }
             }
             case PERIODIC_CONSUMER_ASYNCHRONOUS -> {
                 //