Fix consumer timeout issue after broker restart

wushilin · Jul 25, 2024 · 269537f · 269537f
1 parent 68c0d0c
commit 269537f
Showing 1 changed file with 15 additions and 1 deletion.
diff --git a/...l-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ConsumerManager.java b/...l-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ConsumerManager.java
@@ -94,7 +94,21 @@ public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsetsToSen
         noWakeups++;
         while (inProgress) {
             try {
-                consumer.commitSync(offsetsToSend);
+                while(true) {
+                    try {
+                        consumer.commitSync(offsetsToSend);
+                        // break when offset commit is okay. Do not throw exception to 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");
+                            break;
+                        }
+                    }
+                }
                 inProgress = false;
             } catch (WakeupException w) {
                 log.debug("Got woken up, retry. errors: " + erroneousWakups + " none: " + noWakeups + " correct:" + correctPollWakeups, w);