feature #193: Pause/Resume ParallelConsumer

CHANGELOG.adoc

@@ -12,11 +12,17 @@ endif::[]
 
 == Next Version
 
+== v0.5.1.0
+
+=== Features
+
+* #193: Pause / Resume PC (circuit breaker) without unsubscribing from topics
+
 == v0.5.0.1
 
 === Fixes and Improvements
 
-- fixes: #225 Build support for Java 17, 18 (#289)
+* fixes: #225 Build support for Java 17, 18 (#289)
 
 == v0.5.0.0
 

README.adoc

@@ -250,6 +250,8 @@ without operational burden or harming the cluster's performance
 * Java 8 compatibility
 * Throttle control and broker liveliness management
 * Clean draining shutdown cycle
+* Manual Pause / Resume of entire PC without unsubscribing from topics (useful for implementing a simplistic https://en.wikipedia.org/wiki/Circuit_breaker_design_pattern[circuit breaker])
+** Note: Pausing of a partition is also automatic, whenever back pressure has built up on a given partition
 
 //image:https://codecov.io/gh/astubbs/parallel-consumer/branch/master/graph/badge.svg["Coverage",https://codecov.io/gh/astubbs/parallel-consumer]
 //image:https://travis-ci.com/astubbs/parallel-consumer.svg?branch=master["Build Status", link="https://travis-ci.com/astubbs/parallel-consumer"]
@@ -475,7 +477,6 @@ See {issues_link}/12[issue #12], and the `ParallelConsumer` JavaDoc:
  * @param <K> key consume / produce key type
  * @param <V> value consume / produce value type
  * @see AbstractParallelEoSStreamProcessor
- * @see #poll(Consumer)
  */
 ----
 
@@ -1162,11 +1163,17 @@ endif::[]
 
 == Next Version
 
+== v0.5.1.0
+
+=== Features
+
+* #193: Pause / Resume PC (circuit breaker) without unsubscribing from topics
+
 == v0.5.0.1
 
 === Fixes and Improvements
 
-- fixes: #225 Build support for Java 17, 18 (#289)
+* fixes: #225 Build support for Java 17, 18 (#289)
 
 == v0.5.0.0
 

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

@@ -8,9 +8,9 @@
 import io.confluent.parallelconsumer.internal.DrainingCloseable;
 import lombok.Data;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
 
 import java.util.Collection;
-import java.util.function.Consumer;
 import java.util.regex.Pattern;
 
 // tag::javadoc[]
@@ -24,31 +24,60 @@
  * @param <K> key consume / produce key type
  * @param <V> value consume / produce value type
  * @see AbstractParallelEoSStreamProcessor
- * @see #poll(Consumer)
  */
 // end::javadoc[]
 public interface ParallelConsumer<K, V> extends DrainingCloseable {
 
     /**
-     * @see org.apache.kafka.clients.consumer.KafkaConsumer#subscribe(Collection)
+     * @see KafkaConsumer#subscribe(Collection)
      */
     void subscribe(Collection<String> topics);
 
     /**
-     * @see org.apache.kafka.clients.consumer.KafkaConsumer#subscribe(Pattern)
+     * @see KafkaConsumer#subscribe(Pattern)
      */
     void subscribe(Pattern pattern);
 
     /**
-     * @see org.apache.kafka.clients.consumer.KafkaConsumer#subscribe(Collection, ConsumerRebalanceListener)
+     * @see KafkaConsumer#subscribe(Collection, ConsumerRebalanceListener)
      */
     void subscribe(Collection<String> topics, ConsumerRebalanceListener callback);
 
     /**
-     * @see org.apache.kafka.clients.consumer.KafkaConsumer#subscribe(Pattern, ConsumerRebalanceListener)
+     * @see KafkaConsumer#subscribe(Pattern, ConsumerRebalanceListener)
      */
     void subscribe(Pattern pattern, ConsumerRebalanceListener callback);
 
+    /**
+     * Pause this consumer (i.e. stop processing of messages).
+     * <p>
+     * This operation only has an effect if the consumer is currently running. In all other cases calling this method
+     * will be silent a no-op.
+     * <p>
+     * Once the consumer is paused, the system will stop submitting work to the processing pool. Already submitted in
+     * flight work however will be finished. This includes work that is currently being processed inside a user function
+     * as well as work that has already been submitted to the processing pool but has not been picked up by a free
+     * worker yet.
+     * <p>
+     * General remarks:
+     * <ul>
+     * <li>A paused consumer may still keep polling for new work until internal buffers are filled.</li>
+     * <li>This operation does not actively pause the subscription on the underlying Kafka Broker (compared to
+     * {@link KafkaConsumer#pause KafkaConsumer#pause}).</li>
+     * <li>Pending offset commits will still be performed when the consumer is paused.</li>
+     * </p>
+     */
+    void pauseIfRunning();
+
+    /**
+     * Resume this consumer (i.e. continue processing of messages).
+     * <p>
+     * This operation only has an effect if the consumer is currently paused. In all other cases calling this method
+     * will be a silent no-op.
+     * </p>
+     */
+    void resumeIfPaused();
+
     /**
      * A simple tuple structure.
      *

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

@@ -23,7 +23,7 @@ public class ParallelEoSStreamProcessor<K, V> extends AbstractParallelEoSStreamP
         implements ParallelStreamProcessor<K, V> {
 
     /**
-     * Construct the AsyncConsumer by wrapping this passed in conusmer and producer, which can be configured any which
+     * Construct the AsyncConsumer by wrapping this passed in consumer and producer, which can be configured any which
      * way as per normal.
      *
      * @see ParallelConsumerOptions

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

@@ -6,6 +6,7 @@
 
 import io.confluent.parallelconsumer.state.WorkContainer;
 import lombok.Getter;
+import lombok.ToString;
 import lombok.experimental.Delegate;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
@@ -16,6 +17,7 @@
 /**
  * Internal only view on the {@link PollContext}.
  */
+@ToString
 public class PollContextInternal<K, V> {
 
     @Delegate

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

@@ -6,10 +6,12 @@
 
 import io.confluent.parallelconsumer.state.WorkContainer;
 import lombok.Getter;
+import lombok.ToString;
 
 /**
  * Internal only view of the {@link RecordContext} class.
  */
+@ToString
 public class RecordContextInternal<K, V> {
 
     @Getter

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

@@ -55,6 +55,12 @@ public abstract class AbstractParallelEoSStreamProcessor<K, V> implements Parall
     public static final String MDC_INSTANCE_ID = "pcId";
     public static final String MDC_OFFSET_MARKER = "offset";
 
+    /**
+     * Key for the work container descriptor that will be added to the {@link MDC diagnostic context} while inside a
+     * user function.
+     */
+    private static final String MDC_WORK_CONTAINER_DESCRIPTOR = "offset";
+
     @Getter(PROTECTED)
     protected final ParallelConsumerOptions options;
 
@@ -664,7 +670,7 @@ private <R> void controlLoop(Function<PollContextInternal<K, V>, List<R>> userFu
         log.trace("Loop: Process mailbox");
         processWorkCompleteMailBox();
 
-        if (state == running) {
+        if (isIdlingOrRunning()) {
             // offsets will be committed when the consumer has its partitions revoked
             log.trace("Loop: Maybe commit");
             commitOffsetsMaybe();
@@ -947,9 +953,9 @@ private void processWorkCompleteMailBox() {
                 wm.registerWork(action.getConsumerRecords());
             } else {
                 WorkContainer<K, V> work = action.getWorkContainer();
-                MDC.put(MDC_OFFSET_MARKER, work.toString());
+                MDC.put(MDC_WORK_CONTAINER_DESCRIPTOR, work.toString());
                 wm.handleFutureResult(work);
-                MDC.remove(MDC_OFFSET_MARKER);
+                MDC.remove(MDC_WORK_CONTAINER_DESCRIPTOR);
             }
         }
     }
@@ -984,6 +990,11 @@ private Duration getTimeToBlockFor() {
         return effectiveCommitAttemptDelay;
     }
 
+    private boolean isIdlingOrRunning() {
+        return state == running || state == draining || state == paused;
+    }
+
+
     /**
      * Conditionally commit offsets to broker
      */
@@ -1045,7 +1056,7 @@ private boolean lingeringOnCommitWouldBeBeneficial() {
 
     private Duration getTimeToNextCommitCheck() {
         // draining is a normal running mode for the controller
-        if (state == running || state == draining) {
+        if (isIdlingOrRunning()) {
             Duration timeSinceLastCommit = getTimeSinceLastCheck();
             Duration timeBetweenCommits = getTimeBetweenCommits();
             @SuppressWarnings("UnnecessaryLocalVariable")
@@ -1065,6 +1076,7 @@ private Duration getTimeSinceLastCheck() {
     private void commitOffsetsThatAreReady() {
         log.debug("Committing offsets that are ready...");
         synchronized (commitCommand) {
+            log.debug("Committing offsets that are ready...");
             committer.retrieveOffsetsAndCommit();
             clearCommitCommand();
             this.lastCommitTime = Instant.now();
@@ -1086,7 +1098,7 @@ protected <R> List<ParallelConsumer.Tuple<ConsumerRecord<K, V>, R>> runUserFunct
         try {
             if (log.isDebugEnabled()) {
                 // first offset of the batch
-                MDC.put("offset", workContainerBatch.get(0).offset() + "");
+                MDC.put(MDC_WORK_CONTAINER_DESCRIPTOR, workContainerBatch.get(0).offset() + "");
             }
             log.trace("Pool received: {}", workContainerBatch);
 
@@ -1212,4 +1224,25 @@ private void clearCommitCommand() {
         }
     }
 
+    @Override
+    public void pauseIfRunning() {
+        if (this.state == State.running) {
+            log.info("Transitioning parallel consumer to state paused.");
+            this.state = State.paused;
+        } else {
+            log.debug("Skipping transition of parallel consumer to state paused. Current state is {}.", this.state);
+        }
+    }
+
+    @Override
+    public void resumeIfPaused() {
+        if (this.state == State.paused) {
+            log.info("Transitioning paarallel consumer to state running.");
+            this.state = State.running;
+            notifySomethingToDo();
+        } else {
+            log.debug("Skipping transition of parallel consumer to state running. Current state is {}.", this.state);
+        }
+    }
+
 }

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

@@ -42,6 +42,11 @@ public class BrokerPollSystem<K, V> implements OffsetCommitter {
 
     private Optional<Future<Boolean>> pollControlThreadFuture;
 
+    /**
+     * While {@link io.confluent.parallelconsumer.internal.State#paused paused} is an externally controlled state that
+     * temporarily stops polling and work registration, the {@code paused} flag is used internally to pause
+     * subscriptions if polling needs to be throttled.
+     */
     @Getter
     private volatile boolean paused = false;
 
@@ -325,4 +330,36 @@ public void wakeupIfPaused() {
         if (paused)
             consumerManager.wakeup();
     }
+
+    /**
+     * Pause polling from the underlying Kafka Broker.
+     * <p>
+     * Note: If the poll system is currently not in state {@link io.confluent.parallelconsumer.internal.State#running
+     * running}, calling this method will be a no-op.
+     * </p>
+     */
+    public void pausePollingAndWorkRegistrationIfRunning() {
+        if (this.state == State.running) {
+            log.info("Transitioning broker poll system to state paused.");
+            this.state = State.paused;
+        } else {
+            log.info("Skipping transition of broker poll system to state paused. Current state is {}.", this.state);
+        }
+    }
+
+    /**
+     * Resume polling from the underlying Kafka Broker.
+     * <p>
+     * Note: If the poll system is currently not in state {@link io.confluent.parallelconsumer.internal.State#paused
+     * paused}, calling this method will be a no-op.
+     * </p>
+     */
+    public void resumePollingAndWorkRegistrationIfPaused() {
+        if (this.state == State.paused) {
+            log.info("Transitioning broker poll system to state running.");
+            this.state = State.running;
+        } else {
+            log.info("Skipping transition of broker poll system to state running. Current state is {}.", this.state);
+        }
+    }
 }

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

@@ -1,7 +1,7 @@
 package io.confluent.parallelconsumer.internal;
 
 /*-
- * Copyright (C) 2020-2021 Confluent, Inc.
+ * Copyright (C) 2020-2022 Confluent, Inc.
  */
 
 /**
@@ -10,6 +10,13 @@
 public enum State {
     unused,
     running,
+    /**
+     * When paused, the system will stop submitting work to the processing pool. Polling for new work however may
+     * continue until internal buffers have been filled sufficiently and the auto-throttling takes effect.
+     * In flight work will not be affected by transitioning to this state (i.e. processing will finish without any
+     * interrupts being sent).
+     */
+    paused,
     /**
      * When draining, the system will stop polling for more records, but will attempt to process all already downloaded
      * records. Note that if you choose to close without draining, records already processed will still be committed

parallel-consumer-core/src/test/java/io/confluent/csid/utils/KafkaTestUtils.java

@@ -97,16 +97,19 @@ public void assertCommits(MockProducer mp, List<Integer> expectedOffsets, Option
     }
 
     public List<Integer> getProducerCommitsFlattened(MockProducer mp) {
+        return getProducerCommitsMeta(mp).stream().map(x -> (int) x.offset()).collect(Collectors.toList());
+    }
+
+    public List<OffsetAndMetadata> getProducerCommitsMeta(MockProducer mp) {
         List<Map<String, Map<TopicPartition, OffsetAndMetadata>>> history = mp.consumerGroupOffsetsHistory();
 
-        List<Integer> set = history.stream().flatMap(histories -> {
+        List<OffsetAndMetadata> set = history.stream().flatMap(histories -> {
             // get all partition offsets and flatten
-            var results = new ArrayList<Integer>();
+            ArrayList<OffsetAndMetadata> results = new ArrayList<>();
             var group = histories.get(CONSUMER_GROUP_ID);
             for (var partitionOffsets : group.entrySet()) {
                 OffsetAndMetadata commit = partitionOffsets.getValue();
-                int offset = (int) commit.offset();
-                results.add(offset);
+                results.add(commit);
             }
             return results.stream();
         }).collect(Collectors.toList()); // set - ignore repeated commits ({@link OffsetMap})