fix: Gracefully skip if any encoding can’t be used, fix offset encode drop if too big

Test now exercises the metadata stripping flow.
Capture issue with run length encoding - short cast.

Author: astubbs

CHANGELOG.adoc

@@ -3,7 +3,10 @@
 == v0.2.0.3
 
 * Fixes
-** https://github.com/confluentinc/parallel-consumer/issues/35[Bitset overflow check (#35)] - gracefully drop BitSet encoding as an option if offset different too large
+** https://github.com/confluentinc/parallel-consumer/issues/35[Bitset overflow check (#35)] - gracefully drop BitSet or Runlength encoding as an option if offset difference too large (short overflow)
+*** A new serialisation format will be added in next version - see https://github.com/confluentinc/parallel-consumer/issues/37[Support BitSet encoding lengths longer than Short.MAX_VALUE #37]
+** Gracefully drops encoding attempts if they can't be run
+** Fixes a bug in the offset drop if it can't fit in the offset metadata payload
 
 == v0.2.0.2
 

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

@@ -1,6 +1,6 @@
 package io.confluent.parallelconsumer;
 
-public class BitSetEncodingNotSupportedException extends Exception {
+public class BitSetEncodingNotSupportedException extends EncodingNotSupportedException {
 
     public BitSetEncodingNotSupportedException(String msg) {
         super(msg);

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

@@ -19,7 +19,6 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeoutException;
-import java.util.function.Consumer;
 
 import static io.confluent.csid.utils.BackportUtils.toSeconds;
 import static io.confluent.parallelconsumer.ParallelEoSStreamProcessor.State.closed;
@@ -82,7 +81,7 @@ public void supervise() {
                 try {
                     booleanFuture.get();
                 } catch (Exception e) {
-                    throw new InternalError("Error in " + BrokerPollSystem.class.getSimpleName() + " system.", e);
+                    throw new InternalRuntimeError("Error in " + BrokerPollSystem.class.getSimpleName() + " system.", e);
                 }
             }
         }

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

@@ -0,0 +1,7 @@
+package io.confluent.parallelconsumer;
+
+public class EncodingNotSupportedException extends Exception {
+    public EncodingNotSupportedException(final String message) {
+        super(message);
+    }
+}

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

@@ -0,0 +1,20 @@
+package io.confluent.parallelconsumer;
+
+public class InternalRuntimeError extends RuntimeException {
+
+    public InternalRuntimeError(final String message) {
+        super(message);
+    }
+
+    public InternalRuntimeError(final String message, final Throwable cause) {
+        super(message, cause);
+    }
+
+    public InternalRuntimeError(final Throwable cause) {
+        super(cause);
+    }
+
+    public InternalRuntimeError(final String message, final Throwable cause, final boolean enableSuppression, final boolean writableStackTrace) {
+        super(message, cause, enableSuppression, writableStackTrace);
+    }
+}

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

@@ -0,0 +1,7 @@
+package io.confluent.parallelconsumer;
+
+public class OffsetDecodingError extends Exception {
+    public OffsetDecodingError(final String s, final IllegalArgumentException a) {
+        super(s, a);
+    }
+}

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

@@ -21,7 +21,7 @@ public OffsetEncoder(OffsetSimultaneousEncoder offsetSimultaneousEncoder) {
 
     abstract void encodeCompletedOffset(final int rangeIndex);
 
-    abstract byte[] serialise();
+    abstract byte[] serialise() throws EncodingNotSupportedException;
 
     abstract int getEncodedSize();
 
@@ -33,7 +33,7 @@ byte[] compress() throws IOException {
         return OffsetSimpleSerialisation.compressZstd(this.getEncodedBytes());
     }
 
-    final void register() {
+    void register() throws EncodingNotSupportedException {
         final byte[] bytes = this.serialise();
         final OffsetEncoding encodingType = this.getEncodingType();
         this.register(encodingType, bytes);

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

@@ -12,9 +12,11 @@
 
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.*;
 
 import static io.confluent.csid.utils.Range.range;
+import static io.confluent.csid.utils.StringUtils.msg;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 /**
@@ -44,6 +46,7 @@ public class OffsetMapCodecManager<K, V> {
     public static final int DefaultMaxMetadataSize = 4096;
 
     public static final Charset CHARSET_TO_USE = UTF_8;
+
     private final WorkManager<K, V> wm;
 
     org.apache.kafka.clients.consumer.Consumer<K, V> consumer;
@@ -75,31 +78,40 @@ void loadOffsetMapForPartition(final Set<TopicPartition> assignment) {
             if (offsetAndMeta != null) {
                 long offset = offsetAndMeta.offset();
                 String metadata = offsetAndMeta.metadata();
-                loadOffsetMetadataPayload(offset, tp, metadata);
+                try {
+                    loadOffsetMetadataPayload(offset, tp, metadata);
+                } catch (OffsetDecodingError offsetDecodingError) {
+                    log.error("Error decoding offsets from assigned partition, dropping offset map (will replay previously completed messages - partition: {}, data: {})",
+                            tp, offsetAndMeta, offsetDecodingError);
+                }
             }
         });
     }
 
-    static ParallelConsumer.Tuple<Long, TreeSet<Long>> deserialiseIncompleteOffsetMapFromBase64(long finalBaseComittedOffsetForPartition, String incompleteOffsetMap) {
-        byte[] decode = Base64.getDecoder().decode(incompleteOffsetMap);
+    static ParallelConsumer.Tuple<Long, TreeSet<Long>> deserialiseIncompleteOffsetMapFromBase64(long finalBaseComittedOffsetForPartition, String incompleteOffsetMap) throws OffsetDecodingError {
+        byte[] decode;
+        try {
+            decode = OffsetSimpleSerialisation.decodeBase64(incompleteOffsetMap);
+        } catch (IllegalArgumentException a) {
+            throw new OffsetDecodingError(msg("Error decoding offset metadata, input was: {}", incompleteOffsetMap), a);
+        }
         ParallelConsumer.Tuple<Long, Set<Long>> incompleteOffsets = decodeCompressedOffsets(finalBaseComittedOffsetForPartition, decode);
         TreeSet<Long> longs = new TreeSet<>(incompleteOffsets.getRight());
         return ParallelConsumer.Tuple.pairOf(incompleteOffsets.getLeft(), longs);
     }
 
-    void loadOffsetMetadataPayload(long startOffset, TopicPartition tp, String offsetMetadataPayload) {
+    void loadOffsetMetadataPayload(long startOffset, TopicPartition tp, String offsetMetadataPayload) throws OffsetDecodingError {
         ParallelConsumer.Tuple<Long, TreeSet<Long>> incompletes = deserialiseIncompleteOffsetMapFromBase64(startOffset, offsetMetadataPayload);
         wm.raisePartitionHighWaterMark(incompletes.getLeft(), tp);
         wm.partitionIncompleteOffsets.put(tp, incompletes.getRight());
     }
 
-    String makeOffsetMetadataPayload(long finalOffsetForPartition, TopicPartition tp, Set<Long> incompleteOffsets) {
+    String makeOffsetMetadataPayload(long finalOffsetForPartition, TopicPartition tp, Set<Long> incompleteOffsets) throws EncodingNotSupportedException {
         String offsetMap = serialiseIncompleteOffsetMapToBase64(finalOffsetForPartition, tp, incompleteOffsets);
         return offsetMap;
     }
 
-    @SneakyThrows
-    String serialiseIncompleteOffsetMapToBase64(long finalOffsetForPartition, TopicPartition tp, Set<Long> incompleteOffsets) {
+    String serialiseIncompleteOffsetMapToBase64(long finalOffsetForPartition, TopicPartition tp, Set<Long> incompleteOffsets) throws EncodingNotSupportedException {
         byte[] compressedEncoding = encodeOffsetsCompressed(finalOffsetForPartition, tp, incompleteOffsets);
         String b64 = OffsetSimpleSerialisation.base64(compressedEncoding);
         return b64;
@@ -113,15 +125,16 @@ String serialiseIncompleteOffsetMapToBase64(long finalOffsetForPartition, TopicP
      * <p>
      * Can remove string encoding in favour of the boolean array for the `BitSet` if that's how things settle.
      */
-    @SneakyThrows
-    byte[] encodeOffsetsCompressed(long finalOffsetForPartition, TopicPartition tp, Set<Long> incompleteOffsets) {
+    byte[] encodeOffsetsCompressed(long finalOffsetForPartition, TopicPartition tp, Set<Long> incompleteOffsets) throws EncodingNotSupportedException {
         Long nextExpectedOffset = wm.partitionOffsetHighWaterMarks.get(tp) + 1;
         OffsetSimultaneousEncoder simultaneousEncoder = new OffsetSimultaneousEncoder(finalOffsetForPartition, nextExpectedOffset, incompleteOffsets).invoke();
         if (forcedCodec.isPresent()) {
             OffsetEncoding forcedOffsetEncoding = forcedCodec.get();
-            log.warn("Forcing use of {}", forcedOffsetEncoding);
+            log.warn("Forcing use of {}, for testing", forcedOffsetEncoding);
             Map<OffsetEncoding, byte[]> encodingMap = simultaneousEncoder.getEncodingMap();
             byte[] bytes = encodingMap.get(forcedOffsetEncoding);
+            if (bytes == null)
+                throw new EncodingNotSupportedException(msg("Can't force an encoding that hasn't been run: {}", forcedOffsetEncoding));
             return simultaneousEncoder.packEncoding(new EncodedOffsetPair(forcedOffsetEncoding, ByteBuffer.wrap(bytes)));
         } else {
             return simultaneousEncoder.packSmallest();

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

@@ -67,7 +67,7 @@ class OffsetSimultaneousEncoder {
      * <p>
      * Visible for testing.
      */
-    boolean compressionForced = false;
+    static boolean compressionForced = false;
 
     /**
      * The encoders to run
@@ -169,7 +169,16 @@ public OffsetSimultaneousEncoder invoke() {
     }
 
     private void registerEncodings(final Set<? extends OffsetEncoder> encoders) {
-        encoders.forEach(OffsetEncoder::register);
+        List<OffsetEncoder> toRemove = new ArrayList<>();
+        for (OffsetEncoder encoder : encoders) {
+            try {
+                encoder.register();
+            } catch (EncodingNotSupportedException e) {
+                log.warn("Removing {} encoder, not supported", encoder.getEncodingType().name(), e);
+                toRemove.add(encoder);
+            }
+        }
+        encoders.removeAll(toRemove);
 
         // compressed versions
         // sizes over LARGE_INPUT_MAP_SIZE_THRESHOLD bytes seem to benefit from compression
@@ -184,7 +193,10 @@ private void registerEncodings(final Set<? extends OffsetEncoder> encoders) {
      *
      * @see #packEncoding(EncodedOffsetPair)
      */
-    public byte[] packSmallest() {
+    public byte[] packSmallest() throws EncodingNotSupportedException {
+        if (sortedEncodings.isEmpty()) {
+            throw new EncodingNotSupportedException("No encodings could be used");
+        }
         final EncodedOffsetPair best = this.sortedEncodings.first();
         log.debug("Compression chosen is: {}", best.encoding.name());
         return packEncoding(best);

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

@@ -106,6 +106,9 @@ public enum CommitMode {
      * <p>
      * At the moment this is a sort of sanity check, and was chosen rather arbitriarly. However, one should consider
      * that this is per client, and is a total across all assigned partitions.
+     * <p>
+     * It's important that this is small enough, that you're not at risk of the broker expiring log segments where the
+     * oldest offset resides.
      */
     @Builder.Default
     private final int maxNumberMessagesBeyondBaseCommitOffset = 1000;

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

@@ -234,7 +234,7 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
             wm.onPartitionsRevoked(partitions);
             usersConsumerRebalanceListener.ifPresent(x -> x.onPartitionsRevoked(partitions));
         } catch (Exception e) {
-            throw new InternalError("onPartitionsRevoked event error", e);
+            throw new InternalRuntimeError("onPartitionsRevoked event error", e);
         }
     }
 

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

@@ -7,6 +7,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static io.confluent.csid.utils.StringUtils.msg;
 import static io.confluent.parallelconsumer.OffsetEncoding.RunLength;
 import static io.confluent.parallelconsumer.OffsetEncoding.RunLengthCompressed;
 
@@ -47,13 +48,15 @@ public void encodeCompletedOffset(final int rangeIndex) {
     }
 
     @Override
-    public byte[] serialise() {
+    public byte[] serialise() throws EncodingNotSupportedException {
         runLengthEncodingIntegers.add(currentRunLengthCount.get()); // add tail
 
         ByteBuffer runLengthEncodedByteBuffer = ByteBuffer.allocate(runLengthEncodingIntegers.size() * Short.BYTES);
-        for (final Integer i : runLengthEncodingIntegers) {
-            final short value = i.shortValue();
-            runLengthEncodedByteBuffer.putShort(value);
+        for (final Integer runlength : runLengthEncodingIntegers) {
+            final short shortCastRunlength = runlength.shortValue();
+            if (runlength != shortCastRunlength)
+                throw new RunlengthV1EncodingNotSupported(msg("Runlength too long for Short ({} cast to {})", runlength, shortCastRunlength));
+            runLengthEncodedByteBuffer.putShort(shortCastRunlength);
         }
 
         byte[] array = runLengthEncodedByteBuffer.array();

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

@@ -0,0 +1,7 @@
+package io.confluent.parallelconsumer;
+
+public class RunlengthV1EncodingNotSupported extends EncodingNotSupportedException {
+    public RunlengthV1EncodingNotSupported(final String msg) {
+        super(msg);
+    }
+}

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

@@ -458,10 +458,10 @@ boolean hasCommittableOffsets() {
 
     /**
      * TODO: This entire loop could be possibly redundant, if we instead track low water mark, and incomplete offsets as
-     *  work is submitted and returned.
+     * work is submitted and returned.
+     * <p>
      * todo: refactor into smaller methods?
      */
-    @SneakyThrows
     <R> Map<TopicPartition, OffsetAndMetadata> findCompletedEligibleOffsetsAndRemove(boolean remove) {
         Map<TopicPartition, OffsetAndMetadata> offsetsToSend = new HashMap<>();
         int count = 0;
@@ -526,10 +526,14 @@ <R> Map<TopicPartition, OffsetAndMetadata> findCompletedEligibleOffsetsAndRemove
                 }
 
                 OffsetMapCodecManager<K, V> om = new OffsetMapCodecManager<>(this, this.consumer);
-                String offsetMapPayload = om.makeOffsetMetadataPayload(offsetOfNextExpectedMessage, topicPartitionKey, incompleteOffsets);
-                totalOffsetMetaCharacterLength += offsetMapPayload.length();
-                OffsetAndMetadata offsetWithExtraMap = new OffsetAndMetadata(offsetOfNextExpectedMessage, offsetMapPayload);
-                offsetsToSend.put(topicPartitionKey, offsetWithExtraMap);
+                try {
+                    String offsetMapPayload = om.makeOffsetMetadataPayload(offsetOfNextExpectedMessage, topicPartitionKey, incompleteOffsets);
+                    totalOffsetMetaCharacterLength += offsetMapPayload.length();
+                    OffsetAndMetadata offsetWithExtraMap = new OffsetAndMetadata(offsetOfNextExpectedMessage, offsetMapPayload);
+                    offsetsToSend.put(topicPartitionKey, offsetWithExtraMap);
+                } catch (EncodingNotSupportedException e) {
+                    log.warn("No encodings could be used to encode the offset map, skipping. Warning: messages might be replayed on rebalance", e);
+                }
             }
 
             if (remove) {
@@ -571,9 +575,11 @@ private void maybeStripOffsetPayload(Map<TopicPartition, OffsetAndMetadata> offs
             for (var entry : offsetsToSend.entrySet()) {
                 TopicPartition key = entry.getKey();
                 OffsetAndMetadata v = entry.getValue();
-                OffsetAndMetadata stripped = new OffsetAndMetadata(v.offset(), v.toString());
+                OffsetAndMetadata stripped = new OffsetAndMetadata(v.offset()); // meta data gone
                 offsetsToSend.replace(key, stripped);
             }
+        } else {
+            log.debug("Offset map small enough to fit in payload: {} (max: {})", totalOffsetMetaCharacterLength, OffsetMapCodecManager.DefaultMaxMetadataSize);
         }
     }
 

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

@@ -60,7 +60,6 @@ public class VeryLargeMessageVolumeTest extends BrokerIntegrationTest<String, St
 
 
     @Test
-//    @Disabled("See #35, $37")
     public void shouldNotThrowBitsetTooLongException() {
         runTest(HIGH_MAX_POLL_RECORDS_CONFIG, CommitMode.CONSUMER_ASYNCHRONOUS, ProcessingOrder.UNORDERED);
     }