WIP! feature: Choose between Consumer commit or Producer transactional commits

Reproducing bug 25: confluentinc#25

WIP! Feature: Makes transactions optional

More

More - optional tx test exposes tx state error - non tx works

ReentrantReadWrite lock protects non-thread safe transactional producer from incorrect multithreaded use

Refactor

Fix onClose, remove TransactionState monitoring attempt - not needed

WIP! docs: Adds transaction system architecture

Wider lock to prevent transaction's containing produced messages that they shouldn't

Wider lock to prevent transaction's containing produced messages that they shouldn't

Squash tighten up

Tighten up

Fix

Implement non transactional synchronous commit sync properly

Volume tests adapted to non transactional, passes

More

Test fix - must start tx in MockProducer as well

More

fixed bug in test, added more tests and better verification

Fixes double lock causing deadlock, stops interruption of transaction commits

Fixes wakeup issues with non-transactional commits, adds supervision to poller

Bug: fix example app tests

- Fixes a performance issue with the async committer not being woken up
- Enhances tests to run under multiple commit modes

More

More

WIP! Update tests to run in all 3 commit modes

More

More

More

More

More

- fixes example app tests - incorrectly testing wrong thing and MockProducer not configured to auto complete

Use iteration over stream as can’t workout a concurrent modification here

- remove Optional#ifPresentOrElse} - it’s only @SInCE 9

Fixes vertx app test, adds wiremock

Squash

Squash

Squash

Squash! Fix concurrent modification exception

Squash

bug: Stabilise test and make committer thread revoke partitions and commit

- CloseAndOpenOffset test from some race condition
- Make sure the correct thread owner of the committer is the to close the consumer and thus commit the offsets

Fix concurrency and ProducerFencedExceptions

Enhancement: make the thread responsible for committing the thread which runs the commit upon close
Have onPartitionsRevoked be responsible for committing on close, instead of an explicit call to commit by controller

Make sure Broker Poller now drains properly, committing any waiting work
Add missing revoke flow to MockConsumer wrapper
Add missing latch timeout check

Author: JorgenRingen

.gitignore

@@ -29,4 +29,7 @@ hs_err_pid*
 *.iml
 target/
 .DS_Store
-*.versionsBackup
+*.versionsBackup
+
+# JENV
+.java-version

.idea/runConfigurations/All_examples.xml

@@ -313,18 +313,17 @@ Where `${project.version}` is the version to be used:
 .Setup the client
 [source,java,indent=0]
 ----
-        var options = ParallelConsumerOptions.builder()
-                .ordering(KEY) // <1>
-                .maxConcurrency(1000) // <2>
-                .maxUncommittedMessagesToHandlePerPartition(10000) // <3>
-                .build();
+        ParallelConsumerOptions options = getOptions();
 
         Consumer<String, String> kafkaConsumer = getKafkaConsumer(); // <4>
+        Producer<String, String> kafkaProducer = getKafkaProducer();
+
+        ParallelStreamProcessor<String, String> eosStreamProcessor = ParallelStreamProcessor.createEosStreamProcessor(kafkaConsumer, kafkaProducer, options);
         if (!(kafkaConsumer instanceof MockConsumer)) {
-            kafkaConsumer.subscribe(UniLists.of(inputTopic)); // <5>
+            eosStreamProcessor.subscribe(UniLists.of(inputTopic)); // <5>
         }
 
-        return ParallelStreamProcessor.createEosStreamProcessor(kafkaConsumer, getKafkaProducer(), options);
+        return eosStreamProcessor;
 ----
 <1> Choose your ordering type, `KEY` in this case.
 This ensures maximum concurrency, while ensuring messages are processed and committed in `KEY` order, making sure no offset is committed unless all offsets before it in it's partition, are completed also.
@@ -385,12 +384,17 @@ You can also optionally provide a callback function to be run after the message(
         this.parallelConsumer.pollAndProduce(record -> {
                     var result = processBrokerRecord(record);
                     ProducerRecord<String, String> produceRecord =
-                            new ProducerRecord<>(outputTopic, "a-key", result.payload);
+                            new ProducerRecord<>(outputTopic, record.key(), result.payload);
+
+                    processedCount.incrementAndGet();
                     return UniLists.of(produceRecord);
-                }, consumeProduceResult ->
-                        log.info("Message {} saved to broker at offset {}",
-                                consumeProduceResult.getOut(),
-                                consumeProduceResult.getMeta().offset())
+                }, consumeProduceResult -> {
+                    producedCount.incrementAndGet();
+                    processedAndProducedKeys.add(consumeProduceResult.getIn().key());
+                    log.debug("Message {} saved to broker at offset {}",
+                            consumeProduceResult.getOut(),
+                            consumeProduceResult.getMeta().offset());
+                }
         );
 
 
@@ -412,7 +416,7 @@ In future versions, we plan to look at supporting other streaming systems like h
         var resultStream = parallelConsumer.vertxHttpReqInfoStream(record -> {
             log.info("Concurrently constructing and returning RequestInfo from record: {}", record);
             Map<String, String> params = UniMaps.of("recordKey", record.key(), "payload", record.value());
-            return new RequestInfo("localhost", "/api", params); // <1>
+            return new RequestInfo("localhost", port, "/api", params); // <1>
         });
 ----
 <1> Simply return an object representing the request, the Vert.x HTTP engine will handle the rest, using it's non-blocking engine
@@ -453,7 +457,7 @@ image::https://lucid.app/publicSegments/view/43f2740c-2a7f-4b7f-909e-434a5bbe3fb
     }
 
     void concurrentProcess() {
-        setupConsumer();
+        setupParallelConsumer();
 
         parallelConsumer.poll(record -> {
             log.info("Concurrently processing a record: {}", record);
@@ -562,6 +566,13 @@ your processing, you create side effects in other systems, this pertains to the
 
 CAUTION: This cannot be true for any externally integrated third party system, unless that system is __idempotent__.
 
+You can choose to use the Transactional system, or not.
+
+Pros:
+Cons:
+
+For implementations details, see the <<Transactional System Architecture>> section.
+
 [[streams-usage]]
 == Using with Kafka Streams
 
@@ -748,6 +759,10 @@ Instead of the work thread pool count being the degree of concurrency, it is con
 .Vert.x Architecture
 image::https://lucid.app/publicSegments/view/509df410-5997-46be-98e7-ac7f241780b4/image.png[Vert.x Architecture, align="center"]
 
+=== Transactional System Architecture
+
+image::https://lucid.app/publicSegments/view/7480d948-ed7d-4370-a308-8ec12e6b453b/image.png[]
+
 === Offset Map
 
 ==== Storage

README.adoc

@@ -9,7 +9,11 @@
 public class StringUtils {
 
     public static String msg(String s, Object... args) {
-        String message = MessageFormatter.basicArrayFormat(s, args);
-        return message;
+        return MessageFormatter.basicArrayFormat(s, args);
+    }
+
+    public static boolean isBlank(final String property) {
+        if (property == null) return true;
+        else return property.trim().isEmpty(); // isBlank @since 11
     }
 }

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

@@ -0,0 +1,59 @@
+package io.confluent.parallelconsumer;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.Map;
+
+@Slf4j
+@RequiredArgsConstructor
+public abstract class AbstractOffsetCommitter<K, V> implements OffsetCommitter {
+
+    protected final ConsumerManager<K, V> consumerMgr;
+    protected final WorkManager<K, V> wm;
+
+    /**
+     * Get offsets from {@link WorkManager} that are ready to commit
+     */
+    @Override
+    public void retrieveOffsetsAndCommit() {
+        log.debug("Commit starting - find completed work to commit offsets");
+        // todo shouldn't be removed until commit succeeds (there's no harm in committing the same offset twice)
+        preAcquireWork();
+        try {
+            Map<TopicPartition, OffsetAndMetadata> offsetsToSend = wm.findCompletedEligibleOffsetsAndRemove();
+            if (offsetsToSend.isEmpty()) {
+                log.trace("No offsets ready");
+            } else {
+                log.debug("Will commit offsets for {} partition(s): {}", offsetsToSend.size(), offsetsToSend);
+                ConsumerGroupMetadata groupMetadata = consumerMgr.groupMetadata();
+
+                log.debug("Begin commit");
+                commitOffsets(offsetsToSend, groupMetadata);
+
+                log.debug("On commit success");
+                onOffsetCommitSuccess(offsetsToSend);
+            }
+        } finally {
+            postCommit();
+        }
+    }
+
+    protected void postCommit() {
+        // default noop
+    }
+
+    protected void preAcquireWork() {
+        // default noop
+    }
+
+    private void onOffsetCommitSuccess(final Map<TopicPartition, OffsetAndMetadata> offsetsToSend) {
+        wm.onOffsetCommitSuccess(offsetsToSend);
+    }
+
+    protected abstract void commitOffsets(final Map<TopicPartition, OffsetAndMetadata> offsetsToSend, final ConsumerGroupMetadata groupMetadata);
+
+}