diff --git a/CHANGELOG.adoc b/CHANGELOG.adoc index 14c82dfe20..2c16e0eeda 100644 --- a/CHANGELOG.adoc +++ b/CHANGELOG.adoc @@ -1,5 +1,10 @@ = Change Log +== next + +* Fixes +** https://github.com/confluentinc/parallel-consumer/issues/37[Support BitSet encoding lengths longer than Short.MAX_VALUE #37] - adds new serialisation formats that supports wider range of offsets - (32,767 vs 2,147,483,647) for both BitSet and run-length encoding + == v0.2.0.3 * Fixes diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/BitsetEncoder.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/BitsetEncoder.java index 1961e22be2..c9badd93ea 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/BitsetEncoder.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/BitsetEncoder.java @@ -6,7 +6,7 @@ import java.util.BitSet; import java.util.Optional; -import static io.confluent.parallelconsumer.OffsetEncoding.BitSetCompressed; +import static io.confluent.parallelconsumer.OffsetEncoding.*; /** * Encodes a range of offsets, from an incompletes collection into a BitSet. @@ -30,34 +30,77 @@ */ class BitsetEncoder extends OffsetEncoder { - public static final Short MAX_LENGTH_ENCODABLE = Short.MAX_VALUE; + private final Version version; // default to new version - private final ByteBuffer wrappedBitsetBytesBuffer; + private static final Version DEFAULT_VERSION = Version.v2; + + public static final Integer MAX_LENGTH_ENCODABLE = Integer.MAX_VALUE; + + private ByteBuffer wrappedBitsetBytesBuffer; private final BitSet bitSet; private Optional encodedBytes = Optional.empty(); public BitsetEncoder(int length, OffsetSimultaneousEncoder offsetSimultaneousEncoder) throws BitSetEncodingNotSupportedException { + this(length, offsetSimultaneousEncoder, DEFAULT_VERSION); + } + + public BitsetEncoder(int length, OffsetSimultaneousEncoder offsetSimultaneousEncoder, Version newVersion) throws BitSetEncodingNotSupportedException { super(offsetSimultaneousEncoder); + + this.version = newVersion; + + switch (newVersion) { + case v1 -> initV1(length); + case v2 -> initV2(length); + } + bitSet = new BitSet(length); + } + + /** + * Switch from encoding bitset length as a short to an integer (length of 32,000 was reasonable too short). + *

+ * Integer.MAX_VALUE should always be good enough as system restricts large from being processed at once. + */ + private void initV2(int length) throws BitSetEncodingNotSupportedException { if (length > MAX_LENGTH_ENCODABLE) { // need to upgrade to using Integer for the bitset length, but can't change serialisation format in-place - throw new BitSetEncodingNotSupportedException(StringUtils.msg("Bitset too long to encode, as length overflows Short.MAX_VALUE. Length: {}. (max: {})", length, Short.MAX_VALUE)); + throw new BitSetEncodingNotSupportedException(StringUtils.msg("Bitset V2 too long to encode, as length overflows Integer.MAX_VALUE. Length: {}. (max: {})", length, MAX_LENGTH_ENCODABLE)); + } + // prep bit set buffer + this.wrappedBitsetBytesBuffer = ByteBuffer.allocate(Integer.BYTES + ((length / 8) + 1)); + // bitset doesn't serialise it's set capacity, so we have to as the unused capacity actually means something + this.wrappedBitsetBytesBuffer.putInt(length); + } + + /** + * This was a bit "short" sighted of me.... + */ + private void initV1(int length) throws BitSetEncodingNotSupportedException { + if (length > Short.MAX_VALUE) { + // need to upgrade to using Integer for the bitset length, but can't change serialisation format in-place + throw new BitSetEncodingNotSupportedException("Bitset V1 too long to encode, bitset length overflows Short.MAX_VALUE: " + length + ". (max: " + Short.MAX_VALUE + ")"); } // prep bit set buffer this.wrappedBitsetBytesBuffer = ByteBuffer.allocate(Short.BYTES + ((length / 8) + 1)); // bitset doesn't serialise it's set capacity, so we have to as the unused capacity actually means something this.wrappedBitsetBytesBuffer.putShort((short) length); - bitSet = new BitSet(length); } @Override protected OffsetEncoding getEncodingType() { - return OffsetEncoding.BitSet; + return switch (version) { + case v1 -> BitSet; + case v2 -> BitSetV2; + }; } @Override protected OffsetEncoding getEncodingTypeCompressed() { - return BitSetCompressed; + return switch (version) { + case v1 -> BitSetCompressed; + case v2 -> BitSetV2Compressed; + }; } @Override diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/EncodedOffsetPair.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/EncodedOffsetPair.java index b0d80d8327..58922d3c9e 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/EncodedOffsetPair.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/EncodedOffsetPair.java @@ -14,6 +14,9 @@ import static io.confluent.parallelconsumer.OffsetBitSet.deserialiseBitSetWrap; import static io.confluent.parallelconsumer.OffsetBitSet.deserialiseBitSetWrapToIncompletes; +import static io.confluent.parallelconsumer.OffsetEncoding.*; +import static io.confluent.parallelconsumer.OffsetEncoding.Version.v1; +import static io.confluent.parallelconsumer.OffsetEncoding.Version.v2; import static io.confluent.parallelconsumer.OffsetRunLength.*; import static io.confluent.parallelconsumer.OffsetSimpleSerialisation.decompressZstd; import static io.confluent.parallelconsumer.OffsetSimpleSerialisation.deserialiseByteArrayToBitMapString; @@ -67,7 +70,7 @@ private static byte[] copyBytesOutOfBufferForDebug(ByteBuffer bbData) { static EncodedOffsetPair unwrap(byte[] input) { ByteBuffer wrap = ByteBuffer.wrap(input).asReadOnlyBuffer(); byte magic = wrap.get(); - OffsetEncoding decode = OffsetEncoding.decode(magic); + OffsetEncoding decode = decode(magic); ByteBuffer slice = wrap.slice(); return new EncodedOffsetPair(decode, slice); @@ -78,10 +81,15 @@ public String getDecodedString() { String binaryArrayString = switch (encoding) { case ByteArray -> deserialiseByteArrayToBitMapString(data); case ByteArrayCompressed -> deserialiseByteArrayToBitMapString(decompressZstd(data)); - case BitSet -> deserialiseBitSetWrap(data); - case BitSetCompressed -> deserialiseBitSetWrap(decompressZstd(data)); + case BitSet -> deserialiseBitSetWrap(data, v1); + case BitSetCompressed -> deserialiseBitSetWrap(decompressZstd(data), v1); case RunLength -> runLengthDecodeToString(runLengthDeserialise(data)); case RunLengthCompressed -> runLengthDecodeToString(runLengthDeserialise(decompressZstd(data))); + case BitSetV2-> deserialiseBitSetWrap(data, v2); + case BitSetV2Compressed-> deserialiseBitSetWrap(data, v2); + case RunLengthV2-> deserialiseBitSetWrap(data, v2); + case RunLengthV2Compressed-> deserialiseBitSetWrap(data, v2); + default -> throw new InternalRuntimeError("Invalid state"); // todo why is this needed? what's not covered? }; return binaryArrayString; } @@ -91,11 +99,15 @@ public Tuple> getDecodedIncompletes(long baseOffset) { Tuple> binaryArrayString = switch (encoding) { // case ByteArray -> deserialiseByteArrayToBitMapString(data); // case ByteArrayCompressed -> deserialiseByteArrayToBitMapString(decompressZstd(data)); - case BitSet -> deserialiseBitSetWrapToIncompletes(baseOffset, data); - case BitSetCompressed -> deserialiseBitSetWrapToIncompletes(baseOffset, decompressZstd(data)); - case RunLength -> runLengthDecodeToIncompletes(baseOffset, data); - case RunLengthCompressed -> runLengthDecodeToIncompletes(baseOffset, decompressZstd(data)); - default -> throw new UnsupportedOperationException("Encoding (" + encoding.name() + ") not supported"); + case BitSet -> deserialiseBitSetWrapToIncompletes(encoding, baseOffset, data); + case BitSetCompressed -> deserialiseBitSetWrapToIncompletes(BitSet, baseOffset, decompressZstd(data)); + case RunLength -> runLengthDecodeToIncompletes(encoding, baseOffset, data); + case RunLengthCompressed -> runLengthDecodeToIncompletes(RunLength, baseOffset, decompressZstd(data)); + case BitSetV2 -> deserialiseBitSetWrapToIncompletes(encoding, baseOffset, data); + case BitSetV2Compressed -> deserialiseBitSetWrapToIncompletes(BitSetV2, baseOffset, decompressZstd(data)); + case RunLengthV2 -> runLengthDecodeToIncompletes(encoding, baseOffset, data); + case RunLengthV2Compressed -> runLengthDecodeToIncompletes(RunLengthV2, baseOffset, decompressZstd(data)); + default -> throw new UnsupportedOperationException("Encoding (" + encoding.description() + ") not supported"); }; return binaryArrayString; } diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetBitSet.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetBitSet.java index 25bbc96891..5c68a1bf4f 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetBitSet.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetBitSet.java @@ -15,14 +15,23 @@ import static io.confluent.csid.utils.Range.range; /** + * Deserialisation tools for {@link BitsetEncoder}. + *

+ * todo unify or refactor with {@link BitsetEncoder}. Why was it ever seperate? + * * @see BitsetEncoder */ @Slf4j public class OffsetBitSet { - static String deserialiseBitSetWrap(ByteBuffer wrap) { + static String deserialiseBitSetWrap(ByteBuffer wrap, OffsetEncoding.Version version) { wrap.rewind(); - short originalBitsetSize = wrap.getShort(); + + int originalBitsetSize = switch (version) { + case v1 -> (int)wrap.getShort(); // up cast ok + case v2 -> wrap.getInt(); + }; + ByteBuffer slice = wrap.slice(); return deserialiseBitSet(originalBitsetSize, slice); } @@ -42,9 +51,13 @@ static String deserialiseBitSet(int originalBitsetSize, ByteBuffer s) { return result.toString(); } - static Tuple> deserialiseBitSetWrapToIncompletes(long baseOffset, ByteBuffer wrap) { + static Tuple> deserialiseBitSetWrapToIncompletes(OffsetEncoding encoding, long baseOffset, ByteBuffer wrap) { wrap.rewind(); - short originalBitsetSize = wrap.getShort(); + int originalBitsetSize = switch(encoding) { + case BitSet -> wrap.getShort(); + case BitSetV2 -> wrap.getInt(); + default -> throw new InternalRuntimeError("Invalid state"); + }; ByteBuffer slice = wrap.slice(); Set incompletes = deserialiseBitSetToIncompletes(baseOffset, originalBitsetSize, slice); long highwaterMark = baseOffset + originalBitsetSize; diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetEncoder.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetEncoder.java index 78cb8b7deb..d8a062dd50 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetEncoder.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetEncoder.java @@ -1,10 +1,15 @@ package io.confluent.parallelconsumer; import lombok.SneakyThrows; +import lombok.extern.slf4j.Slf4j; import java.io.IOException; import java.nio.ByteBuffer; +/** + * Base OffsetEncoder + */ +@Slf4j abstract class OffsetEncoder { private final OffsetSimultaneousEncoder offsetSimultaneousEncoder; @@ -40,6 +45,7 @@ void register() throws EncodingNotSupportedException { } private void register(final OffsetEncoding type, final byte[] bytes) { + log.debug("Registering {}, with site {}", type, bytes.length); offsetSimultaneousEncoder.sortedEncodings.add(new EncodedOffsetPair(type, ByteBuffer.wrap(bytes))); offsetSimultaneousEncoder.encodingMap.put(type, bytes); } diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetEncoding.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetEncoding.java index a86e4aaf3d..94853d8c91 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetEncoding.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetEncoding.java @@ -13,15 +13,34 @@ import java.util.function.Function; import java.util.stream.Collectors; +import static io.confluent.parallelconsumer.OffsetEncoding.Version.v1; +import static io.confluent.parallelconsumer.OffsetEncoding.Version.v2; + @ToString @RequiredArgsConstructor enum OffsetEncoding { - ByteArray((byte) 'L'), - ByteArrayCompressed((byte) 'î'), - BitSet((byte) 'l'), - BitSetCompressed((byte) 'a'), - RunLength((byte) 'n'), - RunLengthCompressed((byte) 'J'); + ByteArray(v1, (byte) 'L'), + ByteArrayCompressed(v1, (byte) 'î'), + BitSet(v1, (byte) 'l'), + BitSetCompressed(v1, (byte) 'a'), + RunLength(v1, (byte) 'n'), + RunLengthCompressed(v1, (byte) 'J'), + /** + * switch from encoding bitset length as a short to an integer (length of 32,000 was reasonable too short) + */ + BitSetV2(v2, (byte) 'o'), + BitSetV2Compressed(v2, (byte) 's'), + /** + * switch from encoding run lengths as Shorts to Integers + */ + RunLengthV2(v2, (byte) 'e'), + RunLengthV2Compressed(v2, (byte) 'p'); + + enum Version { + v1, v2 + } + + public final Version version; @Getter public final byte magicByte; @@ -36,4 +55,8 @@ public static OffsetEncoding decode(byte magic) { return encoding; } } + + public String description() { + return name() + ":" + version; + } } diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetMapCodecManager.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetMapCodecManager.java index 0d47acc705..28e3ec934c 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetMapCodecManager.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetMapCodecManager.java @@ -89,13 +89,13 @@ void loadOffsetMapForPartition(final Set assignment) { } static ParallelConsumer.Tuple> deserialiseIncompleteOffsetMapFromBase64(long finalBaseComittedOffsetForPartition, String incompleteOffsetMap) throws OffsetDecodingError { - byte[] decode; + byte[] decodedBytes; try { - decode = OffsetSimpleSerialisation.decodeBase64(incompleteOffsetMap); + decodedBytes = OffsetSimpleSerialisation.decodeBase64(incompleteOffsetMap); } catch (IllegalArgumentException a) { throw new OffsetDecodingError(msg("Error decoding offset metadata, input was: {}", incompleteOffsetMap), a); } - ParallelConsumer.Tuple> incompleteOffsets = decodeCompressedOffsets(finalBaseComittedOffsetForPartition, decode); + ParallelConsumer.Tuple> incompleteOffsets = decodeCompressedOffsets(finalBaseComittedOffsetForPartition, decodedBytes); TreeSet longs = new TreeSet<>(incompleteOffsets.getRight()); return ParallelConsumer.Tuple.pairOf(incompleteOffsets.getLeft(), longs); } @@ -147,13 +147,13 @@ byte[] encodeOffsetsCompressed(long finalOffsetForPartition, TopicPartition tp, * * @return Set of offsets which are not complete. */ - static ParallelConsumer.Tuple> decodeCompressedOffsets(long finalOffsetForPartition, byte[] s) { - if (s.length == 0) { + static ParallelConsumer.Tuple> decodeCompressedOffsets(long finalOffsetForPartition, byte[] decodedBytes) { + if (decodedBytes.length == 0) { // no offset bitmap data return ParallelConsumer.Tuple.pairOf(finalOffsetForPartition, UniSets.of()); } - EncodedOffsetPair result = EncodedOffsetPair.unwrap(s); + EncodedOffsetPair result = EncodedOffsetPair.unwrap(decodedBytes); ParallelConsumer.Tuple> incompletesTuple = result.getDecodedIncompletes(finalOffsetForPartition); diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetRunLength.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetRunLength.java index 4de3bfe851..aab3d10ff6 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetRunLength.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetRunLength.java @@ -8,14 +8,22 @@ import lombok.experimental.UtilityClass; import lombok.extern.slf4j.Slf4j; +import java.nio.BufferUnderflowException; import java.nio.ByteBuffer; +import java.nio.IntBuffer; import java.nio.ShortBuffer; import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Set; +import java.util.concurrent.Callable; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import java.util.function.Supplier; + +import static io.confluent.parallelconsumer.OffsetEncoding.BitSet; +import static io.confluent.parallelconsumer.OffsetEncoding.BitSetV2; @Slf4j @UtilityClass @@ -69,47 +77,70 @@ static String runLengthDecodeToString(final List in) { /** * @see #runLengthEncode */ - static Tuple> runLengthDecodeToIncompletes(final long baseOffset, final ByteBuffer in) { + static Tuple> runLengthDecodeToIncompletes(OffsetEncoding encoding, final long baseOffset, final ByteBuffer in) { in.rewind(); - final ShortBuffer shortBuffer = in.asShortBuffer(); + final ShortBuffer v1ShortBuffer = in.asShortBuffer(); + final IntBuffer v2IntegerBuffer = in.asIntBuffer(); final var incompletes = new HashSet(1); // we don't know the capacity yet long highestWatermarkSeen = 0L; + Supplier hasRemainingTest = () -> { + return switch (encoding.version) { + case v1 -> v1ShortBuffer.hasRemaining(); + case v2 -> v2IntegerBuffer.hasRemaining(); + }; + }; if (log.isTraceEnabled()) { - var shorts = new ArrayList(); - while (shortBuffer.hasRemaining()) { - shorts.add(shortBuffer.get()); + // print out all run lengths + var runlengths = new ArrayList(); + try { + while (hasRemainingTest.get()) { + Number runLength = switch (encoding.version) { + case v1 -> v1ShortBuffer.get(); + case v2 -> v2IntegerBuffer.get(); + }; + runlengths.add(runLength); + } + } catch (BufferUnderflowException u) { + log.error("Error decoding offsets", u); } - log.debug("Unrolled shorts: {}", shorts); - shortBuffer.rewind(); + log.debug("Unrolled runlengths: {}", runlengths); + v1ShortBuffer.rewind(); + v2IntegerBuffer.rewind(); } - // + // decodes incompletes boolean currentRunlengthIsComplete = false; long currentOffset = baseOffset; - while (shortBuffer.hasRemaining()) { - short runLength = shortBuffer.get(); - highestWatermarkSeen = currentOffset + runLength; - if (currentRunlengthIsComplete) { - log.trace("Ignoring {} completed offset", runLength); - currentOffset += runLength; - } else { - log.trace("Adding {} incomplete offset", runLength); - for (int relativeOffset = 0; relativeOffset < runLength; relativeOffset++) { - incompletes.add(currentOffset); - currentOffset++; + while (hasRemainingTest.get()) { + try { + Number runLength = switch (encoding.version) { + case v1 -> v1ShortBuffer.get(); + case v2 -> v2IntegerBuffer.get(); + }; + highestWatermarkSeen = currentOffset + runLength.longValue(); + + if (currentRunlengthIsComplete) { + log.trace("Ignoring {} completed offset", runLength); + currentOffset += runLength.longValue(); + } else { + log.trace("Adding {} incomplete offset", runLength); + for (int relativeOffset = 0; relativeOffset < runLength.longValue(); relativeOffset++) { + incompletes.add(currentOffset); + currentOffset++; + } } + } catch (BufferUnderflowException u) { + log.error("Error decoding offsets", u); + throw u; } currentRunlengthIsComplete = !currentRunlengthIsComplete; // toggle } return Tuple.pairOf(highestWatermarkSeen, incompletes); } - /** - * TODO: Is there an advantage to keeping it as an list of Shorts instead of Integer? - */ static List runLengthDeserialise(final ByteBuffer in) { // view as short buffer in.rewind(); diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetSimultaneousEncoder.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetSimultaneousEncoder.java index cbbd4e7abd..f8f933a153 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetSimultaneousEncoder.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetSimultaneousEncoder.java @@ -11,6 +11,8 @@ import java.util.*; import static io.confluent.csid.utils.Range.range; +import static io.confluent.parallelconsumer.OffsetEncoding.Version.v1; +import static io.confluent.parallelconsumer.OffsetEncoding.Version.v2; /** * Encode with multiple strategies at the same time. @@ -23,7 +25,8 @@ class OffsetSimultaneousEncoder { /** - * Size threshold in bytes after which compressing the encodings will be compared + * Size threshold in bytes after which compressing the encodings will be compared, as it seems to be typically worth + * the extra compression step when beyond this size in the source array. */ public static final int LARGE_INPUT_MAP_SIZE_THRESHOLD = 200; @@ -60,7 +63,8 @@ class OffsetSimultaneousEncoder { * @see #packSmallest() */ @Getter - TreeSet sortedEncodings = new TreeSet<>(); + PriorityQueue sortedEncodings = new PriorityQueue(); + /** * Force the encoder to also add the compressed versions. Useful for testing. @@ -79,7 +83,10 @@ public OffsetSimultaneousEncoder(long lowWaterMark, Long nextExpectedOffset, Set this.nextExpectedOffset = nextExpectedOffset; this.incompleteOffsets = incompleteOffsets; - length = (int) (this.nextExpectedOffset - this.lowWaterMark); + long longLength = this.nextExpectedOffset - this.lowWaterMark; + length = (int) longLength; + // sanity + if (longLength != length) throw new IllegalArgumentException("Integer overflow"); initEncoders(); } @@ -90,13 +97,19 @@ private void initEncoders() { } try { - BitsetEncoder bitsetEncoder = new BitsetEncoder(length, this); - encoders.add(bitsetEncoder); + encoders.add(new BitsetEncoder(length, this, v1)); + } catch (BitSetEncodingNotSupportedException a) { + log.warn("Cannot use {} encoder ({})", BitsetEncoder.class.getSimpleName(), a.getMessage()); + } + + try { + encoders.add(new BitsetEncoder(length, this, v2)); } catch (BitSetEncodingNotSupportedException a) { - log.warn("Cannot use {} encoder", BitsetEncoder.class.getSimpleName(), a); + log.warn("Cannot use {} encoder ({})", BitsetEncoder.class.getSimpleName(), a.getMessage()); } - encoders.add(new RunLengthEncoder(this)); + encoders.add(new RunLengthEncoder(this, v1)); + encoders.add(new RunLengthEncoder(this, v2)); } /** @@ -174,7 +187,7 @@ private void registerEncodings(final Set encoders) { try { encoder.register(); } catch (EncodingNotSupportedException e) { - log.warn("Removing {} encoder, not supported", encoder.getEncodingType().name(), e); + log.warn("Removing {} encoder, not supported ({})", encoder.getEncodingType().description(), e.getMessage()); toRemove.add(encoder); } } @@ -197,7 +210,7 @@ public byte[] packSmallest() throws EncodingNotSupportedException { if (sortedEncodings.isEmpty()) { throw new EncodingNotSupportedException("No encodings could be used"); } - final EncodedOffsetPair best = this.sortedEncodings.first(); + final EncodedOffsetPair best = this.sortedEncodings.poll(); log.debug("Compression chosen is: {}", best.encoding.name()); return packEncoding(best); } diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RunLengthEncoder.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RunLengthEncoder.java index 43dd6065eb..438b950167 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RunLengthEncoder.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RunLengthEncoder.java @@ -8,33 +8,42 @@ 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; +import static io.confluent.parallelconsumer.OffsetEncoding.*; class RunLengthEncoder extends OffsetEncoder { - private final AtomicInteger currentRunLengthCount; - private final AtomicBoolean previousRunLengthState; + private int currentRunLengthCount = 0; + private boolean previousRunLengthState = false; + private final List runLengthEncodingIntegers; private Optional encodedBytes = Optional.empty(); - public RunLengthEncoder(OffsetSimultaneousEncoder offsetSimultaneousEncoder) { + private final Version version; // default to new version + + private static final Version DEFAULT_VERSION = Version.v2; + + public RunLengthEncoder(OffsetSimultaneousEncoder offsetSimultaneousEncoder, Version newVersion) { super(offsetSimultaneousEncoder); // run length setup - currentRunLengthCount = new AtomicInteger(); - previousRunLengthState = new AtomicBoolean(false); runLengthEncodingIntegers = new ArrayList<>(); + version = newVersion; } @Override protected OffsetEncoding getEncodingType() { - return RunLength; + return switch (version) { + case v1 -> RunLength; + case v2 -> RunLengthV2; + }; } @Override protected OffsetEncoding getEncodingTypeCompressed() { - return RunLengthCompressed; + return switch (version) { + case v1 -> RunLengthCompressed; + case v2 -> RunLengthV2Compressed; + }; } @Override @@ -49,14 +58,26 @@ public void encodeCompletedOffset(final int rangeIndex) { @Override public byte[] serialise() throws EncodingNotSupportedException { - runLengthEncodingIntegers.add(currentRunLengthCount.get()); // add tail + runLengthEncodingIntegers.add(currentRunLengthCount); // add tail + + int entryWidth = switch (version) { + case v1 -> Short.BYTES; + case v2 -> Integer.BYTES; + }; + ByteBuffer runLengthEncodedByteBuffer = ByteBuffer.allocate(runLengthEncodingIntegers.size() * entryWidth); - ByteBuffer runLengthEncodedByteBuffer = ByteBuffer.allocate(runLengthEncodingIntegers.size() * Short.BYTES); 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); + switch (version) { + case v1 -> { + final short shortCastRunlength = runlength.shortValue(); + if (runlength != shortCastRunlength) + throw new RunlengthV1EncodingNotSupported(msg("Runlength too long for Short ({} cast to {})", runlength, shortCastRunlength)); + runLengthEncodedByteBuffer.putShort(shortCastRunlength); + } + case v2 -> { + runLengthEncodedByteBuffer.putInt(runlength); + } + } } byte[] array = runLengthEncodedByteBuffer.array(); @@ -76,13 +97,13 @@ protected byte[] getEncodedBytes() { private void encodeRunLength(final boolean currentIsComplete) { // run length - boolean currentOffsetMatchesOurRunLengthState = previousRunLengthState.get() == currentIsComplete; + boolean currentOffsetMatchesOurRunLengthState = previousRunLengthState == currentIsComplete; if (currentOffsetMatchesOurRunLengthState) { - currentRunLengthCount.getAndIncrement(); + currentRunLengthCount++; } else { - previousRunLengthState.set(currentIsComplete); - runLengthEncodingIntegers.add(currentRunLengthCount.get()); - currentRunLengthCount.set(1); // reset to 1 + previousRunLengthState = currentIsComplete; + runLengthEncodingIntegers.add(currentRunLengthCount); + currentRunLengthCount = 1; // reset to 1 } } } diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/OffsetEncodingTests.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/OffsetEncodingTests.java index 254a26d311..b51a9f6119 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/OffsetEncodingTests.java +++ b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/OffsetEncodingTests.java @@ -10,17 +10,17 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.ValueSource; -import org.junitpioneer.jupiter.CartesianProductTest; -import org.junitpioneer.jupiter.CartesianValueSource; import pl.tlinkowski.unij.api.UniSets; import java.nio.ByteBuffer; import java.util.*; import java.util.concurrent.BrokenBarrierException; import java.util.concurrent.FutureTask; +import java.util.stream.Collectors; import static io.confluent.parallelconsumer.OffsetEncoding.*; -import static org.assertj.core.api.Assertions.*; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.hamcrest.Matchers.in; import static org.hamcrest.Matchers.not; import static org.junit.Assume.assumeThat; @@ -76,22 +76,11 @@ void largeIncompleteOffsetValues(long nextExpectedOffset) { Map encodingMap = encoder.getEncodingMap(); // - byte[] smallestBytes = null; - if (nextExpectedOffset > Short.MAX_VALUE) { - assertThatThrownBy(() -> encoder.packSmallest()).hasMessageContaining("No encodings"); - return; - } else { - smallestBytes = encoder.packSmallest(); - } + byte[] smallestBytes = encoder.packSmallest(); EncodedOffsetPair unwrap = EncodedOffsetPair.unwrap(smallestBytes); ParallelConsumer.Tuple> decodedIncompletes = unwrap.getDecodedIncompletes(lowWaterMark); assertThat(decodedIncompletes.getRight()).containsExactlyInAnyOrderElementsOf(incompletes); - if (nextExpectedOffset - lowWaterMark > BitsetEncoder.MAX_LENGTH_ENCODABLE) - assertThat(encodingMap.keySet()).as("Gracefully ignores that Bitset can't be supported").doesNotContain(OffsetEncoding.BitSet); - else - assertThat(encodingMap.keySet()).contains(OffsetEncoding.BitSet); - // for (OffsetEncoding encodingToUse : OffsetEncoding.values()) { log.info("Testing {}", encodingToUse); @@ -114,6 +103,7 @@ void largeIncompleteOffsetValues(long nextExpectedOffset) { * * @see #ensureEncodingGracefullyWorksWhenOffsetsArentSequentialTwo */ + @SneakyThrows @ParameterizedTest @EnumSource(OffsetEncoding.class) void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(OffsetEncoding encoding) throws BrokenBarrierException, InterruptedException { @@ -130,13 +120,20 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset records.add(new ConsumerRecord<>(INPUT_TOPIC, 0, 5, "akey", "avalue")); records.add(new ConsumerRecord<>(INPUT_TOPIC, 0, 69, "akey", "avalue")); // will complete records.add(new ConsumerRecord<>(INPUT_TOPIC, 0, 100, "akey", "avalue")); + records.add(new ConsumerRecord<>(INPUT_TOPIC, 0, 1_000, "akey", "avalue")); records.add(new ConsumerRecord<>(INPUT_TOPIC, 0, 20_000, "akey", "avalue")); // near upper limit of Short.MAX_VALUE - records.add(new ConsumerRecord<>(INPUT_TOPIC, 0, 25_000, "akey", "avalue")); // near upper limit of Short.MAX_VALUE + records.add(new ConsumerRecord<>(INPUT_TOPIC, 0, 25_000, "akey", "avalue")); // will complete, near upper limit of Short.MAX_VALUE records.add(new ConsumerRecord<>(INPUT_TOPIC, 0, 30_000, "akey", "avalue")); // near upper limit of Short.MAX_VALUE - // No encoders support this yet - until https://github.com/confluentinc/parallel-consumer/issues/37: - // records.add(new ConsumerRecord<>(INPUT_TOPIC, 0, 40_000, "akey", "avalue")); // higher than Short.MAX_VALUE - // records.add(new ConsumerRecord<>(INPUT_TOPIC, 0, 40_000 + Short.MAX_VALUE, "akey", "avalue")); // runlength higher than Short.MAX_VALUE + // Extremely large tests for v2 encoders + records.add(new ConsumerRecord<>(INPUT_TOPIC, 0, 40_000, "akey", "avalue")); // higher than Short.MAX_VALUE + int avoidOffByOne = 2; + records.add(new ConsumerRecord<>(INPUT_TOPIC, 0, 40_000 + Short.MAX_VALUE + avoidOffByOne, "akey", "avalue")); // runlength higher than Short.MAX_VALUE + + var firstSucceededRecordRemoved = new ArrayList<>(records); + firstSucceededRecordRemoved.remove(firstSucceededRecordRemoved.stream().filter(x -> x.offset() == 0).findFirst().get()); + firstSucceededRecordRemoved.remove(firstSucceededRecordRemoved.stream().filter(x -> x.offset() == 69).findFirst().get()); + firstSucceededRecordRemoved.remove(firstSucceededRecordRemoved.stream().filter(x -> x.offset() == 25_000).findFirst().get()); // ktu.send(consumerSpy, records); @@ -148,7 +145,7 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset objectObjectHashMap.put(tp, records); ConsumerRecords crs = new ConsumerRecords<>(objectObjectHashMap); - // + // write offsets Map completedEligibleOffsetsAndRemove; { WorkManager wmm = new WorkManager<>(options, consumerSpy); @@ -163,28 +160,36 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset completeWork(wmm, work, 25_000); completedEligibleOffsetsAndRemove = wmm.findCompletedEligibleOffsetsAndRemove(); + + // check for graceful fall back to the smallest available encoder + OffsetMapCodecManager om = new OffsetMapCodecManager<>(wmm, consumerSpy); + Set collect = firstSucceededRecordRemoved.stream().map(x -> x.offset()).collect(Collectors.toSet()); + OffsetMapCodecManager.forcedCodec = Optional.empty(); // turn off forced + String bestPayload = om.makeOffsetMetadataPayload(1, tp, collect); + assertThat(bestPayload).isNotEmpty(); } consumerSpy.commitSync(completedEligibleOffsetsAndRemove); + // read offsets { WorkManager newWm = new WorkManager<>(options, consumerSpy); newWm.onPartitionsAssigned(UniSets.of(tp)); newWm.registerWork(crs); List> workContainers = newWm.maybeGetWork(); - var firstSucceededRecordRemoved = new ArrayList<>(records); - firstSucceededRecordRemoved.remove(firstSucceededRecordRemoved.stream().filter(x -> x.offset() == 0).findFirst().get()); - firstSucceededRecordRemoved.remove(firstSucceededRecordRemoved.stream().filter(x -> x.offset() == 69).findFirst().get()); - firstSucceededRecordRemoved.remove(firstSucceededRecordRemoved.stream().filter(x -> x.offset() == 25_000).findFirst().get()); - if (encoding.equals(BitSet)) { - // plain bitset is too large to be used, offsets are dropped - assertThatThrownBy(() -> - assertThat(workContainers).extracting(WorkContainer::getCr) - .containsExactlyElementsOf(firstSucceededRecordRemoved)) - .hasMessageContaining("but some elements were not expected") - .hasMessageContaining("offset = 25000"); - - } else { - assertThat(workContainers).extracting(WorkContainer::getCr).containsExactlyElementsOf(firstSucceededRecordRemoved); + switch (encoding) { + case BitSet, BitSetCompressed, // BitSetV1 both get a short overflow due to the length being too long + BitSetV2, // BitSetv2 uncompressed is too large to fit in metadata payload + RunLength, RunLengthCompressed // RunLength V1 max runlength is Short.MAX_VALUE + -> { + assertThatThrownBy(() -> + assertThat(workContainers).extracting(WorkContainer::getCr) + .containsExactlyElementsOf(firstSucceededRecordRemoved)) + .hasMessageContaining("but some elements were not expected") + .hasMessageContaining("offset = 25000"); + } + default -> { + assertThat(workContainers).extracting(WorkContainer::getCr).containsExactlyElementsOf(firstSucceededRecordRemoved); + } } } } diff --git a/parallel-consumer-core/src/test/resources/junit-platform.properties b/parallel-consumer-core/src/test/resources/junit-platform.properties index e3c240beb9..7f40a1ce1e 100644 --- a/parallel-consumer-core/src/test/resources/junit-platform.properties +++ b/parallel-consumer-core/src/test/resources/junit-platform.properties @@ -2,4 +2,4 @@ # Copyright (C) 2020 Confluent, Inc. # -#junit.jupiter.displayname.generator.default = io.confluent.csid.utils.ReplaceCamelCase \ No newline at end of file +junit.jupiter.displayname.generator.default = io.confluent.csid.utils.ReplaceCamelCase \ No newline at end of file diff --git a/parallel-consumer-core/src/test/resources/logback-test.xml b/parallel-consumer-core/src/test/resources/logback-test.xml index f99d18a072..0bc43b4706 100644 --- a/parallel-consumer-core/src/test/resources/logback-test.xml +++ b/parallel-consumer-core/src/test/resources/logback-test.xml @@ -28,14 +28,14 @@ - - - - - - + + + + + + - + @@ -45,15 +45,16 @@ + - - - + + +