Add option to specify timeout for how long to wait offset commits in …

…periodic-consumer-sync commit-mode

Timeouts on commits will cause the control-loop to exit.

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

@@ -169,6 +169,13 @@ public enum CommitMode {
     @Builder.Default
     private final Duration sendTimeout = Duration.ofSeconds(10);
 
+    /**
+     * Controls how long to block while waiting for offsets to be committed.
+     * Only relevant if using {@link CommitMode#PERIODIC_CONSUMER_SYNC} commit-mode.
+     */
+    @Builder.Default
+    private final Duration offsetCommitTimeout = Duration.ofSeconds(10);
+
     public void validate() {
         Objects.requireNonNull(consumer, "A consumer must be supplied");
 

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

@@ -42,6 +42,8 @@ public class ConsumerOffsetCommitter<K, V> extends AbstractOffsetCommitter<K, V>
 
     private final CommitMode commitMode;
 
+    private final Duration commitTimeout;
+
     private Optional<Thread> owningThread = Optional.empty();
 
     /**
@@ -57,6 +59,7 @@ public class ConsumerOffsetCommitter<K, V> extends AbstractOffsetCommitter<K, V>
     public ConsumerOffsetCommitter(final ConsumerManager<K, V> newConsumer, final WorkManager<K, V> newWorkManager, final ParallelConsumerOptions options) {
         super(newConsumer, newWorkManager);
         commitMode = options.getCommitMode();
+        commitTimeout = options.getOffsetCommitTimeout();
         if (commitMode.equals(PERIODIC_TRANSACTIONAL_PRODUCER)) {
             throw new IllegalArgumentException("Cannot use " + commitMode + " when using " + this.getClass().getSimpleName());
         }
@@ -151,7 +154,7 @@ private void commitAndWait() {
             try {
                 log.debug("Waiting on a commit response");
                 Duration timeout = AbstractParallelEoSStreamProcessor.DEFAULT_TIMEOUT;
-                CommitResponse take = commitResponseQueue.poll(timeout.toMillis(), TimeUnit.MILLISECONDS); // blocks, drain until we find our response
+                CommitResponse take = commitResponseQueue.poll(commitTimeout.toMillis(), TimeUnit.MILLISECONDS); // blocks, drain until we find our response
                 if (take == null)
                     throw new InternalRuntimeError(msg("Timeout waiting for commit response {} to request {}", timeout, commitRequest));
                 waitingOnCommitResponse = take.getRequest().getId() != commitRequest.getId();