fixes #409: Adds support for compacted topics and commit offset resetting

CHANGELOG.adoc

@@ -14,6 +14,14 @@ ifndef::github_name[]
 toc::[]
 endif::[]
 
+== 0.5.2.4
+
+=== Fixes
+
+* fixes #409: Adds support for compacted topics and commit offset resetting (#425)
+** Truncate the offset state when bootstrap polled offset higher or lower than committed
+** Prune missing records from the tracked incomplete offset state, when they're missing from polled batches
+
 == 0.5.2.3
 
 === Improvements

README.adoc

@@ -1287,6 +1287,14 @@ ifndef::github_name[]
 toc::[]
 endif::[]
 
+== 0.5.2.4
+
+=== Fixes
+
+* fixes #409: Adds support for compacted topics and commit offset resetting (#425)
+** Truncate the offset state when bootstrap polled offset higher or lower than committed
+** Prune missing records from the tracked incomplete offset state, when they're missing from polled batches
+
 == 0.5.2.3
 
 === Improvements

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

@@ -8,21 +8,23 @@
 import lombok.experimental.UtilityClass;
 
 import java.time.Duration;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
+import java.util.*;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static java.time.Duration.ofMillis;
 
 @UtilityClass
 public class JavaUtils {
 
-    public static <T> Optional<T> getLast(final List<T> commitHistory) {
-        if (commitHistory.isEmpty()) return Optional.empty();
-        return Optional.of(commitHistory.get(commitHistory.size() - 1));
+    public static <T> Optional<T> getLast(final List<T> someList) {
+        if (someList.isEmpty()) return Optional.empty();
+        return Optional.of(someList.get(someList.size() - 1));
+    }
+
+    public static <T> Optional<T> getFirst(final List<T> someList) {
+        return someList.isEmpty() ? Optional.empty() : Optional.of(someList.get(0));
     }
 
     public static <T> Optional<T> getOnlyOne(final Map<String, T> stringMapMap) {
@@ -56,4 +58,19 @@ public static <K, V1, V2> Map<K, V2> remap(Map<K, V1> map,
                 ));
     }
 
+    public static List<String> getRandom(List<String> list, int quantity) {
+        if (list.size() < quantity) {
+            throw new IllegalArgumentException("List size is less than quantity");
+        }
+
+        return createRandomIntStream(list.size())
+                .limit(quantity)
+                .map(list::get)
+                .collect(Collectors.toList());
+    }
+
+    private static Stream<Integer> createRandomIntStream(int range) {
+        final Random random = new Random();
+        return Stream.generate(() -> random.nextInt(range));
+    }
 }

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

@@ -12,6 +12,7 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.common.annotation.InterfaceStability;
 
 import java.time.Duration;
 import java.util.Objects;
@@ -31,6 +32,7 @@
 @Builder(toBuilder = true)
 @ToString
 @FieldNameConstants
+@InterfaceStability.Evolving
 public class ParallelConsumerOptions<K, V> {
 
     /**

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

@@ -244,19 +244,19 @@ protected AbstractParallelEoSStreamProcessor(ParallelConsumerOptions<K, V> newOp
     protected AbstractParallelEoSStreamProcessor(ParallelConsumerOptions<K, V> newOptions, PCModule<K, V> module) {
         Objects.requireNonNull(newOptions, "Options must be supplied");
 
+        options = newOptions;
+        this.consumer = options.getConsumer();
+
+        validateConfiguration();
+
         module.setParallelEoSStreamProcessor(this);
 
-        log.info("Confluent Parallel Consumer initialise... Options: {}", newOptions);
+        log.info("Confluent Parallel Consumer initialise... groupId: {}, Options: {}",
+                newOptions.getConsumer().groupMetadata().groupId(),
+                newOptions);
 
-        options = newOptions;
-        options.validate();
 
         this.dynamicExtraLoadFactor = module.dynamicExtraLoadFactor();
-        this.consumer = options.getConsumer();
-
-        checkGroupIdConfigured(consumer);
-        checkNotSubscribed(consumer);
-        checkAutoCommitIsDisabled(consumer);
 
         workerThreadPool = setupWorkerPool(newOptions.getMaxConcurrency());
 
@@ -276,6 +276,14 @@ protected AbstractParallelEoSStreamProcessor(ParallelConsumerOptions<K, V> newOp
         }
     }
 
+    private void validateConfiguration() {
+        options.validate();
+
+        checkGroupIdConfigured(consumer);
+        checkNotSubscribed(consumer);
+        checkAutoCommitIsDisabled(consumer);
+    }
+
     private void checkGroupIdConfigured(final org.apache.kafka.clients.consumer.Consumer<K, V> consumer) {
         try {
             consumer.groupMetadata();

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

@@ -27,7 +27,7 @@ public EpochAndRecordsMap(ConsumerRecords<K, V> poll, PartitionStateManager<K, V
         poll.partitions().forEach(partition -> {
             var records = poll.records(partition);
             Long epochOfPartition = pm.getEpochOfPartition(partition);
-            RecordsAndEpoch entry = new RecordsAndEpoch(epochOfPartition, records);
+            RecordsAndEpoch entry = new RecordsAndEpoch(partition, epochOfPartition, records);
             recordMap.put(partition, entry);
         });
     }
@@ -63,6 +63,7 @@ public int count() {
 
     @Value
     public class RecordsAndEpoch {
+        @NonNull TopicPartition topicPartition;
         @NonNull Long epochOfPartitionAtPoll;
         @NonNull List<ConsumerRecord<K, V>> records;
     }

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

@@ -6,6 +6,7 @@
 
 import io.confluent.csid.utils.StringUtils;
 import io.confluent.parallelconsumer.internal.InternalRuntimeException;
+import io.confluent.parallelconsumer.state.PartitionState;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
@@ -27,12 +28,13 @@
  * Sequential or not. Because as records are always in commit order, if we've seen a range of offsets, we know we've
  * seen all that exist (within said range). So if offset 8 is missing from the partition, we will encode it as having
  * been completed (when in fact it doesn't exist), because we only compare against known incompletes, and assume all
- * others are complete.
+ * others are complete. See {@link PartitionState#incompleteOffsets} for more discussion on this.
  * <p>
  * So, when we deserialize, the INCOMPLETES collection is then restored, and that's what's used to compare to see if a
  * record should be skipped or not. So if record 8 is recorded as completed, it will be absent from the restored
  * INCOMPLETES list, and we are assured we will never see record 8.
  *
+ * @see PartitionState#incompleteOffsets
  * @see RunLengthEncoder
  * @see OffsetBitSet
  */

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

@@ -3,6 +3,7 @@
 /*-
  * Copyright (C) 2020-2022 Confluent, Inc.
  */
+
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.ToString;
@@ -15,6 +16,11 @@
 import static io.confluent.parallelconsumer.offsets.OffsetEncoding.Version.v1;
 import static io.confluent.parallelconsumer.offsets.OffsetEncoding.Version.v2;
 
+/**
+ * Offset encoding MagicNumbers to {@link OffsetEncoder}.
+ *
+ * @author Antony Stubbs
+ */
 @ToString
 @RequiredArgsConstructor
 public enum OffsetEncoding {

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

@@ -72,6 +72,7 @@ public static class HighestOffsetAndIncompletes {
         /**
          * Of the offsets encoded, the incomplete ones.
          */
+        // todo change to List as Sets have no order
         Set<Long> incompleteOffsets;
 
         public static HighestOffsetAndIncompletes of(long highestSeenOffset) {
@@ -164,7 +165,7 @@ public static HighestOffsetAndIncompletes deserialiseIncompleteOffsetMapFromBase
     PartitionState<K, V> decodePartitionState(TopicPartition tp, OffsetAndMetadata offsetData) throws OffsetDecodingError {
         HighestOffsetAndIncompletes incompletes = deserialiseIncompleteOffsetMapFromBase64(offsetData);
         log.debug("Loaded incomplete offsets from offset payload {}", incompletes);
-        return new PartitionState<K, V>(tp, incompletes);
+        return new PartitionState<>(tp, incompletes);
     }
 
     public String makeOffsetMetadataPayload(long baseOffsetForPartition, PartitionState<K, V> state) throws NoEncodingPossibleException {