fixes #329: Committing around transaction markers causes encoder to crash

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

@@ -31,6 +31,7 @@ static <KK, VV> ParallelStreamProcessor<KK, VV> createEosStreamProcessor(Paralle
      *
      * @param usersVoidConsumptionFunction the function
      */
+    // todo why isn't this in ParallelConsumer ?
     void poll(Consumer<PollContext<K, V>> usersVoidConsumptionFunction);
 
 

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

@@ -297,7 +297,7 @@ private void checkNotSubscribed(org.apache.kafka.clients.consumer.Consumer<K, V>
         Set<String> subscription = consumerToCheck.subscription();
         Set<TopicPartition> assignment = consumerToCheck.assignment();
         if (!subscription.isEmpty() || !assignment.isEmpty()) {
-            throw new IllegalStateException("Consumer subscription must be managed by this class. Use " + this.getClass().getName() + "#subcribe methods instead.");
+            throw new IllegalStateException("Consumer subscription must be managed by the Parallel Consumer. Use " + this.getClass().getName() + "#subcribe methods instead.");
         }
     }
 
@@ -1158,7 +1158,7 @@ protected void addToMailbox(WorkContainer<K, V> wc) {
     }
 
     public void registerWork(EpochAndRecordsMap<K, V> polledRecords) {
-        log.debug("Adding {} to mailbox...", polledRecords);
+        log.trace("Adding {} to mailbox...", polledRecords);
         workMailBox.add(ControllerEventMessage.of(polledRecords));
     }
 

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

@@ -167,13 +167,13 @@ PartitionState<K, V> decodePartitionState(TopicPartition tp, OffsetAndMetadata o
         return new PartitionState<K, V>(tp, incompletes);
     }
 
-    public String makeOffsetMetadataPayload(long finalOffsetForPartition, PartitionState<K, V> state) throws NoEncodingPossibleException {
-        String offsetMap = serialiseIncompleteOffsetMapToBase64(finalOffsetForPartition, state);
+    public String makeOffsetMetadataPayload(long baseOffsetForPartition, PartitionState<K, V> state) throws NoEncodingPossibleException {
+        String offsetMap = serialiseIncompleteOffsetMapToBase64(baseOffsetForPartition, state);
         return offsetMap;
     }
 
-    String serialiseIncompleteOffsetMapToBase64(long finalOffsetForPartition, PartitionState<K, V> state) throws NoEncodingPossibleException {
-        byte[] compressedEncoding = encodeOffsetsCompressed(finalOffsetForPartition, state);
+    String serialiseIncompleteOffsetMapToBase64(long baseOffsetForPartition, PartitionState<K, V> state) throws NoEncodingPossibleException {
+        byte[] compressedEncoding = encodeOffsetsCompressed(baseOffsetForPartition, state);
         String b64 = OffsetSimpleSerialisation.base64(compressedEncoding);
         return b64;
     }
@@ -186,7 +186,7 @@ String serialiseIncompleteOffsetMapToBase64(long finalOffsetForPartition, Partit
      * <p>
      * Can remove string encoding in favour of the boolean array for the `BitSet` if that's how things settle.
      */
-    byte[] encodeOffsetsCompressed(long finalOffsetForPartition, PartitionState<K, V> partitionState) throws NoEncodingPossibleException {
+    byte[] encodeOffsetsCompressed(long baseOffsetForPartition, PartitionState<K, V> partitionState) throws NoEncodingPossibleException {
         var incompleteOffsets = partitionState.getIncompleteOffsetsBelowHighestSucceeded();
         long highestSucceeded = partitionState.getOffsetHighestSucceeded();
         if (log.isDebugEnabled()) {
@@ -195,7 +195,7 @@ byte[] encodeOffsetsCompressed(long finalOffsetForPartition, PartitionState<K, V
                     highestSucceeded,
                     partitionState.getIncompleteOffsetsBelowHighestSucceeded());
         }
-        OffsetSimultaneousEncoder simultaneousEncoder = new OffsetSimultaneousEncoder(finalOffsetForPartition, highestSucceeded, incompleteOffsets).invoke();
+        OffsetSimultaneousEncoder simultaneousEncoder = new OffsetSimultaneousEncoder(baseOffsetForPartition, highestSucceeded, incompleteOffsets).invoke();
 
         //
         if (forcedCodec.isPresent()) {

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

@@ -4,6 +4,8 @@
  * Copyright (C) 2020-2022 Confluent, Inc.
  */
 
+import io.confluent.parallelconsumer.ParallelConsumerOptions;
+import io.confluent.parallelconsumer.state.PartitionState;
 import io.confluent.parallelconsumer.state.WorkManager;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -12,6 +14,7 @@
 import java.util.*;
 
 import static io.confluent.csid.utils.Range.range;
+import static io.confluent.csid.utils.StringUtils.msg;
 import static io.confluent.parallelconsumer.offsets.OffsetEncoding.Version.v1;
 import static io.confluent.parallelconsumer.offsets.OffsetEncoding.Version.v2;
 
@@ -81,18 +84,21 @@ public class OffsetSimultaneousEncoder {
      */
     private final Set<OffsetEncoder> encoders;
 
-    public OffsetSimultaneousEncoder(long lowWaterMark, long highestSucceededOffset, Set<Long> incompleteOffsets) {
-        this.lowWaterMark = lowWaterMark;
+    public OffsetSimultaneousEncoder(long baseOffsetToCommit, long highestSucceededOffset, Set<Long> incompleteOffsets) {
+        this.lowWaterMark = baseOffsetToCommit;
         this.incompleteOffsets = incompleteOffsets;
 
         //
         if (highestSucceededOffset == -1) { // nothing succeeded yet
-            highestSucceededOffset = lowWaterMark;
+            highestSucceededOffset = baseOffsetToCommit;
         }
 
+        highestSucceededOffset = maybeRaiseOffsetHighestSucceeded(baseOffsetToCommit, highestSucceededOffset);
+
         long bitsetLengthL = highestSucceededOffset - this.lowWaterMark + 1;
         if (bitsetLengthL < 0) {
-            throw new IllegalStateException("Cannot have negative length BitSet");
+            throw new IllegalStateException(msg("Cannot have negative length BitSet (calculated length: {}, base offset to commit: {}, highest succeeded offset: {})",
+                    bitsetLengthL, baseOffsetToCommit, highestSucceededOffset));
         }
 
         // BitSet only support Integer.MAX_VALUE bits
@@ -103,6 +109,35 @@ public OffsetSimultaneousEncoder(long lowWaterMark, long highestSucceededOffset,
         this.encoders = initEncoders();
     }
 
+    /**
+     * Ensure that the {@param #highestSucceededOffset} is always at least a single offset behind the {}@param
+     * baseOffsetToCommit}. Needed to allow us to jump over gaps in the partitions such as transaction markers.
+     * <p>
+     * Under normal operation, it is expected that the highest succeeded offset will generally always be higher than the
+     * next expected offset to poll. This is because PC processes records well beyond the
+     * {@link PartitionState#getOffsetHighestSequentialSucceeded()} all the time, unless operation in
+     * {@link ParallelConsumerOptions.ProcessingOrder#PARTITION} order. So this situation - where the highest succeeded
+     * offset is below the next offset to poll at the time of commit - will either be an incredibly rare case: only at
+     * the very beginning of processing records, or where ALL records are slow enough or blocked, or in synthetically
+     * created scenarios (like test cases).
+     */
+    private long maybeRaiseOffsetHighestSucceeded(long baseOffsetToCommit, long highestSucceededOffset) {
+        long nextExpectedMinusOne = baseOffsetToCommit - 1;
+
+        boolean gapLargerThanOne = highestSucceededOffset < nextExpectedMinusOne;
+        if (gapLargerThanOne) {
+            long gap = nextExpectedMinusOne - highestSucceededOffset;
+            log.debug("Gap detected in partition (highest succeeded: {} while next expected poll offset: {} - gap is {}), probably tx markers. Moving highest succeeded to next expected - 1",
+                    highestSucceededOffset,
+                    nextExpectedMinusOne,
+                    gap);
+            // jump straight to the lowest incomplete - 1, allows us to jump over gaps in the partitions such as transaction markers
+            highestSucceededOffset = nextExpectedMinusOne;
+        }
+
+        return highestSucceededOffset;
+    }
+
     private Set<OffsetEncoder> initEncoders() {
         var newEncoders = new HashSet<OffsetEncoder>();
         if (length > LARGE_INPUT_MAP_SIZE_THRESHOLD) {

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

@@ -80,6 +80,9 @@ public class PartitionState<K, V> {
 
     /**
      * Highest offset which has completed successfully ("succeeded").
+     * <p>
+     * Note that this may in some conditions, there may be a gap between this and the next offset to poll - that being,
+     * there may be some number of transaction marker records above it, and the next offset to poll.
      */
     @Getter(PUBLIC)
     private long offsetHighestSucceeded = KAFKA_OFFSET_ABSENCE;
@@ -133,7 +136,7 @@ private void maybeRaiseHighestSeenOffset(final long offset) {
         }
     }
 
-    public void onOffsetCommitSuccess(final OffsetAndMetadata committed) {
+    public void onOffsetCommitSuccess(OffsetAndMetadata committed) { //NOSONAR
         setClean();
     }
 
@@ -167,7 +170,7 @@ public int getCommitQueueSize() {
     public void onSuccess(WorkContainer<K, V> work) {
         long offset = work.offset();
 
-        WorkContainer<K, V> removedFromQueue = this.commitQueue.remove(work.offset());
+        WorkContainer<K, V> removedFromQueue = this.commitQueue.remove(offset);
         assert (removedFromQueue != null);
 
         boolean removedFromIncompletes = this.incompleteOffsets.remove(offset);
@@ -195,9 +198,10 @@ private void updateHighestSucceededOffsetSoFar(WorkContainer<K, V> work) {
     }
 
     public void addWorkContainer(WorkContainer<K, V> wc) {
-        maybeRaiseHighestSeenOffset(wc.offset());
-        commitQueue.put(wc.offset(), wc);
-        incompleteOffsets.add(wc.offset());
+        long newOffset = wc.offset();
+        maybeRaiseHighestSeenOffset(newOffset);
+        commitQueue.put(newOffset, wc);
+        incompleteOffsets.add(newOffset);
     }
 
     /**
@@ -224,6 +228,9 @@ private OffsetAndMetadata createOffsetAndMetadata() {
                 .orElseGet(() -> new OffsetAndMetadata(nextOffset));
     }
 
+    /**
+     * Defines as the offset one below the highest sequentially succeeded offset
+     */
     private long getNextExpectedPolledOffset() {
         return getOffsetHighestSequentialSucceeded() + 1;
     }
@@ -249,6 +256,10 @@ public Set<Long> getIncompleteOffsetsBelowHighestSucceeded() {
                 .collect(Collectors.toSet()));
     }
 
+    /**
+     * Defined for our purpose (as only used in definition of what offset to poll for next), as the offset one below the
+     * lowest incomplete offset.
+     */
     public long getOffsetHighestSequentialSucceeded() {
         if (this.incompleteOffsets.isEmpty()) {
             return this.offsetHighestSeen;

parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/BrokerIntegrationTest.java

@@ -9,6 +9,7 @@
 
 import io.confluent.csid.testcontainers.FilteredTestContainerSlf4jLogConsumer;
 import io.confluent.parallelconsumer.integrationTests.utils.KafkaClientUtils;
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.CreateTopicsResult;
 import org.apache.kafka.clients.admin.NewTopic;
@@ -53,6 +54,7 @@ public abstract class BrokerIntegrationTest<K, V> {
         kafkaContainer.start();
     }
 
+    @Getter
     protected KafkaClientUtils kcu = new KafkaClientUtils(kafkaContainer);
 
     @BeforeAll