feature #520: Thread safe User API exposure of ALL Consumer APIs

CHANGELOG.adoc

@@ -16,6 +16,18 @@ endif::[]
 
 == 0.6.0.0
 
+=== Features
+
+TODO changelog and readme docs
+
+* feature #520: Thread safe User API exposure of ALL Consumer APIs #346
+** fixes #191: Need ability to seek to the beginning of a partition
+** fixes #186: Ensure all PC API's are thread safe
+*** E.g. let's you use all consumer APIs, including seek, end offsets, lag, assignment, paused, even poll, commit and pause
+*** Both a restricted interface and a full implementation are provided
+*** Allows PC to be used by any client expecting a Kafka Consumer interface, but enables use of high speed PC APIs enabling no head of line blocking and other features aside from the concurrent processing ones.
+See <<docs>> section for elaboration
+
 === Improvements
 
 * major: New IPC system using lightweight Lambda actor queue #325

README.adoc

@@ -243,6 +243,9 @@ without operational burden or harming the cluster's performance
 * Offsets committed correctly, in order, of only processed messages, regardless of concurrency level or retries
 * Batch support in all versions of the API to process batches of messages in parallel instead of single messages.
 ** Particularly useful for when your processing function can work with more than a single record at a time - e.g. sending records to an API which has a batch version like Elasticsearch
+
+== TODO consumer interface documentation
+
 * Vert.x and Reactor.io non-blocking library integration
 ** Non-blocking I/O work management
 ** Vert.x's WebClient and general Vert.x Future support
@@ -472,6 +475,7 @@ See {issues_link}/12[issue #12], and the `ParallelConsumer` JavaDoc:
 
 [source,java]
 ----
+
 /**
  * Asynchronous / concurrent message consumer for Kafka.
  * <p>
@@ -1301,6 +1305,18 @@ endif::[]
 
 == 0.6.0.0
 
+=== Features
+
+TODO changelog and readme docs
+
+* feature #520: Thread safe User API exposure of ALL Consumer APIs #346
+** fixes #191: Need ability to seek to the beginning of a partition
+** fixes #186: Ensure all PC API's are thread safe
+*** E.g. let's you use all consumer APIs, including seek, end offsets, lag, assignment, paused, even poll, commit and pause
+*** Both a restricted interface and a full implementation are provided
+*** Allows PC to be used by any client expecting a Kafka Consumer interface, but enables use of high speed PC APIs enabling no head of line blocking and other features aside from the concurrent processing ones.
+See <<docs>> section for elaboration
+
 === Improvements
 
 * major: New IPC system using lightweight Lambda actor queue #325

parallel-consumer-core/src/main/java/io/confluent/csid/actors/ActorImpl.java

@@ -142,7 +142,7 @@ private <R> CompletableFuture<R> waitForState(ActorState targetState) {
         }
 
         var future = new CompletableFuture<R>();
-        log.debug("Waiting for state {} to be reached", targetState);
+        log.debug("Waiting on actor {} for state {} to be reached", getActorName(), targetState);
         var timer = Time.SYSTEM.timer(stateTimeout);
         while (!state.get().equals(targetState)) {
             stateLock.lock();

parallel-consumer-core/src/main/java/io/confluent/csid/utils/JavaUtils.java

@@ -79,4 +79,7 @@ private static Stream<Integer> createRandomIntStream(int range) {
         return Collectors.toCollection(TreeSet::new);
     }
 
+    public static boolean isEmpty(Collection<?> collection) {
+        return collection == null || collection.isEmpty();
+    }
 }

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

@@ -0,0 +1,59 @@
+package io.confluent.parallelconsumer;
+
+/*-
+ * Copyright (C) 2020-2022 Confluent, Inc.
+ */
+
+import io.confluent.parallelconsumer.internal.ConsumerFacadeForPC;
+import io.confluent.parallelconsumer.internal.ConsumerFacadeStrictImpl;
+import io.confluent.parallelconsumer.internal.FullConsumerFacadeImpl;
+import io.confluent.parallelconsumer.internal.PCConsumerAPIStrict;
+import org.apache.kafka.clients.consumer.Consumer;
+
+/**
+ * Get access to the underlying {@link Consumer} interface.
+ *
+ * @author Antony Stubbs
+ * @see FullConsumerFacadeImpl
+ * @see PCConsumerAPIStrict
+ * @see ConsumerFacadeStrictImpl
+ */
+public interface ConsumerApiAccess<K, V> {
+
+    /**
+     * All consumer style methods, including ones that can't run. For which, you can either choose to no-op or throw an
+     * exception.
+     * todo docs
+     * <p>
+     * An implementation of the Kafka Consumer API, with unsupported functions which throw exceptions. Useful for when you
+     * want to let something else use PC, where it's unaware of it's capabilities, yet you know it won't call unsupported
+     * methods.
+     *
+     * @see FullConsumerFacadeImpl
+     * @see #partialStrictKafkaConsumer()
+     */
+    FullConsumerFacadeImpl<K, V> fullConsumerFacade();
+
+    /**
+     * The {@link PCConsumerAPIStrict} APIs, plus a set of implementations of {@link Consumer}, which are implemented,
+     * but not in the strict contract of the Kafka Consumer API.
+     *
+     * @see ConsumerFacadeForPC
+     */
+    // todo def needs a better name
+    ConsumerFacadeForPC consumerFacadeForPC();
+
+    /**
+     * Partial implementation of {@link Consumer}, for which methods are implemented in teh strict sense of the
+     * contract.
+     * <p>
+     * Use this if possible instead of {@link #fullConsumerFacade()}, as it only has the functions that are allowed to
+     * be called.
+     * <p>
+     * Use {@link #fullConsumerFacade()} if you need something that "is a"
+     * {@link org.apache.kafka.clients.consumer.Consumer}, but won't have unsupported functions called.
+     *
+     * @see PCConsumerAPIStrict
+     */
+    PCConsumerAPIStrict partialStrictKafkaConsumer();
+}

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

@@ -14,6 +14,7 @@
 import java.util.regex.Pattern;
 
 // tag::javadoc[]
+
 /**
  * Asynchronous / concurrent message consumer for Kafka.
  * <p>
@@ -89,6 +90,13 @@ public interface ParallelConsumer<K, V> extends DrainingCloseable {
      */
     void resumeIfPaused();
 
+    /**
+     * APIs to directly control the consumer through various {@link org.apache.kafka.clients.consumer.Consumer} methods.
+     * PC will react appropriately, with respect to what type of method is called. See the method documentation for
+     * specifics for each method you call - they all work slightly differently from each other.
+     */
+    ConsumerApiAccess<K, V> consumerApiAccess();
+
     /**
      * A simple tuple structure.
      *
@@ -97,7 +105,9 @@ public interface ParallelConsumer<K, V> extends DrainingCloseable {
      */
     @Data
     class Tuple<L, R> {
+
         private final L left;
+
         private final R right;
 
         public static <LL, RR> Tuple<LL, RR> pairOf(LL l, RR r) {

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

@@ -59,6 +59,16 @@ public abstract class AbstractParallelEoSStreamProcessor<K, V> extends Rebalance
 
     public static final String MDC_INSTANCE_ID = "pcId";
 
+    /**
+     * @see #consumerApiAccess()
+     */
+    private final ConsumerApiAccess<K, V> consumerFacade;
+
+    @Override
+    public ConsumerApiAccess<K, V> consumerApiAccess() {
+        return consumerFacade;
+    }
+
     /**
      * Key for the work container descriptor that will be added to the {@link MDC diagnostic context} while inside a
      * user function.
@@ -236,6 +246,8 @@ protected AbstractParallelEoSStreamProcessor(ParallelConsumerOptions<K, V> newOp
 
         this.brokerPollSubsystem = module.brokerPoller(this);
 
+        this.consumerFacade = new ConsumerAccessImpl<>(brokerPollSubsystem, this);
+
         this.rebalanceHandler = module.parallelEoSStreamProcessor;
 
         if (options.isProducerSupplied()) {
@@ -299,13 +311,12 @@ private void checkNotSubscribed(org.apache.kafka.clients.consumer.Consumer<K, V>
         }
     }
 
-    // replace with facade delegate
     @Override
     public void subscribe(String topic) {
         subscribe(Collections.singletonList(topic));
     }
 
-    // replace with facade delegate - in controller refactor branch
+    // replace with facade delegate
     @Override
     public void subscribe(Collection<String> topics) {
         log.debug("Subscribing to {}", topics);
@@ -1070,7 +1081,7 @@ private Duration getTimeSinceLastCheck() {
      * Visible for testing
      */
     protected void commitOffsetsThatAreReady() throws InterruptedException, TimeoutException {
-        if (state.equals(UNUSED)) {
+        if (state.equals(UNUSED) || !wm.isDirty()) {
             return;
         }
 

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

@@ -122,10 +122,11 @@ private boolean controlLoop() throws TimeoutException, InterruptedException {
                 handlePoll();
 
                 // old: - would check if it was asked
-//                var commitData = wm.collectCommitDataForDirtyPartitions();
-//                maybeDoCommit(commitData);
+                // var commitData = wm.collectCommitDataForDirtyPartitions();
+                // maybeDoCommit(commitData);
 
                 getMyActor().process();
+                committer.ifPresent(kvConsumerOffsetCommitter -> kvConsumerOffsetCommitter.getMyActor().process());
 
                 switch (runState) {
                     case DRAINING -> {
@@ -332,7 +333,7 @@ public void retrieveOffsetsAndCommit(CommitData offsetsToCommit) {
         }
     }
 
-    // removed as the commiter will do the commit directly if instructed through messaging
+    // removed as the committer will do the commit directly if instructed through messaging
 //    /**
 //     * Will silently skip if not configured with a committer
 //     */

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

@@ -0,0 +1,38 @@
+package io.confluent.parallelconsumer.internal;
+
+/*-
+ * Copyright (C) 2020-2022 Confluent, Inc.
+ */
+
+import io.confluent.parallelconsumer.ConsumerApiAccess;
+import io.confluent.parallelconsumer.ParallelConsumer;
+import lombok.Value;
+
+/**
+ * @author Antony Stubbs
+ * @see ConsumerApiAccess
+ */
+@Value
+public class ConsumerAccessImpl<K, V> implements ConsumerApiAccess<K, V> {
+
+    BrokerPollSystem<K, V> poller;
+
+    ParallelConsumer<K, V> pcApi;
+
+    @Override
+    public FullConsumerFacadeImpl<K, V> fullConsumerFacade() {
+        // new or reused?
+        return new FullConsumerFacadeImpl<>(pcApi, poller);
+    }
+
+    @Override
+    public ConsumerFacadeForPC consumerFacadeForPC() {
+        return new ConsumerFacadeForPCImpl<>(pcApi, poller);
+    }
+
+    @Override
+    public PCConsumerAPIStrict partialStrictKafkaConsumer() {
+        // new or reused?
+        return new ConsumerFacadeStrictImpl<>(poller);
+    }
+}

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

@@ -0,0 +1,51 @@
+package io.confluent.parallelconsumer.internal;
+
+/*-
+ * Copyright (C) 2020-2022 Confluent, Inc.
+ */
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.TopicPartition;
+
+import java.time.Duration;
+import java.util.Collection;
+
+/**
+ * The set of implementations of {@link Consumer}, which are implemented, but not in the strict contract of the Kafka
+ * Consumer API.
+ * <p>
+ * Typically, this involves interpreting the intent, and making the equivalent call to PC.
+ *
+ * @author Antony Stubbs
+ * @see PCConsumerAPIStrict
+ */
+// todo def needs a better name
+// todo should this extend PCConsumerAPIStrict or not? If not, should there be another interface which does?
+public interface ConsumerFacadeForPC extends PCConsumerAPIStrict {
+
+    /**
+     * Like {@link Consumer#pause}, except it pauses ALL PARTITIONS through the PC controller. After enough back
+     * pressure is built up, the underlying consumer will pause.
+     * <p>
+     * Note: this implementation cannot pause individual partitions, only ALL partitions. The parameter is ignored.
+     *
+     * @param collection ignored
+     */
+    void pause(Collection<TopicPartition> collection);
+
+    /**
+     * Like {@link Consumer#resume}, except it resumes ALL PARTITIONS through the PC controller. After enough back
+     * pressure is released, the underlying consumer will resume.
+     * <p>
+     * Note: this implementation cannot pause individual partitions, only ALL partitions. The parameter is ignored.
+     *
+     * @param collection ignored
+     */
+    void resume(Collection<TopicPartition> collection);
+
+
+    void close();
+
+    void close(final Duration timeout);
+
+}

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

@@ -0,0 +1,63 @@
+package io.confluent.parallelconsumer.internal;
+
+/*-
+ * Copyright (C) 2020-2022 Confluent, Inc.
+ */
+
+import io.confluent.parallelconsumer.ParallelConsumer;
+import lombok.experimental.FieldDefaults;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.TopicPartition;
+
+import java.time.Duration;
+import java.util.Collection;
+
+import static io.confluent.csid.utils.JavaUtils.isEmpty;
+import static lombok.AccessLevel.PRIVATE;
+
+/**
+ * @author Antony Stubbs
+ * @see ConsumerFacadeForPC
+ */
+@Slf4j
+@ThreadSafe
+@FieldDefaults(makeFinal = true, level = PRIVATE)
+public class ConsumerFacadeForPCImpl<K, V> extends ConsumerFacadeStrictImpl<K, V> implements ConsumerFacadeForPC {
+
+    ParallelConsumer<K, V> pcApi;
+
+    public ConsumerFacadeForPCImpl(ParallelConsumer<K, V> pcApi, BrokerPollSystem<K, V> poller) {
+        super(poller);
+        this.pcApi = pcApi;
+    }
+
+    @Override
+    public void pause(final Collection<TopicPartition> collection) {
+        warnInvalidUse(collection);
+        pcApi.pauseIfRunning();
+    }
+
+    private static void warnInvalidUse(Collection<?> collection) {
+        if (!isEmpty(collection)) {
+            log.warn("Pausing specific partitions not supported yet, pausing all");
+        }
+    }
+
+    @Override
+    public void resume(Collection<TopicPartition> collection) {
+        warnInvalidUse(collection);
+        pcApi.resumeIfPaused();
+    }
+
+    // option to have this shutdown PC?
+    @Override
+    public void close() {
+        throw new UnsupportedOperationException("Not supported yet.");
+    }
+
+    @Override
+    public void close(final Duration timeout) {
+        throw new UnsupportedOperationException("Not supported yet.");
+    }
+
+}