diff --git a/parallel-consumer-core/pom.xml b/parallel-consumer-core/pom.xml index a0afe5800..ed9edf6e1 100644 --- a/parallel-consumer-core/pom.xml +++ b/parallel-consumer-core/pom.xml @@ -20,6 +20,7 @@ + org.apache.kafka kafka-clients @@ -37,6 +38,7 @@ 1.1.8.4 compile + org.awaitility @@ -81,6 +83,7 @@ org.testcontainers postgresql + test org.postgresql @@ -103,20 +106,18 @@ truth-java8-extension test - - com.google.flogger - flogger - - - com.google.flogger - flogger-slf4j-backend - org.threeten threeten-extra 1.7.0 test + + one.util + streamex + 0.8.1 + test + diff --git a/parallel-consumer-core/src/main/java/io/confluent/csid/utils/JavaUtils.java b/parallel-consumer-core/src/main/java/io/confluent/csid/utils/JavaUtils.java index 8515733eb..bba786ed7 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/csid/utils/JavaUtils.java +++ b/parallel-consumer-core/src/main/java/io/confluent/csid/utils/JavaUtils.java @@ -12,6 +12,8 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.function.Function; +import java.util.stream.Collectors; import static java.time.Duration.ofMillis; @@ -39,4 +41,19 @@ public static boolean isGreaterThan(Duration compare, Duration to) { return compare.compareTo(to) > 0; } + /** + * A shortcut for changing only the values of a Map. + *

+ * https://stackoverflow.com/a/50740570/105741 + */ + public static Map remap(Map map, + Function function) { + return map.entrySet() + .stream() // or parallel + .collect(Collectors.toMap( + Map.Entry::getKey, + e -> function.apply(e.getValue()) + )); + } + } diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ErrorInUserFunctionException.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ErrorInUserFunctionException.java index 00c77ef23..bd1b81f36 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ErrorInUserFunctionException.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ErrorInUserFunctionException.java @@ -1,13 +1,13 @@ package io.confluent.parallelconsumer; /*- - * Copyright (C) 2020-2021 Confluent, Inc. + * Copyright (C) 2020-2022 Confluent, Inc. */ /** * This exception is only used when there is an exception thrown from code provided by the user. */ -public class ErrorInUserFunctionException extends RuntimeException { +public class ErrorInUserFunctionException extends ParallelConsumerException { public ErrorInUserFunctionException(final String message, final Throwable cause) { super(message, cause); } diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelConsumerException.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelConsumerException.java new file mode 100644 index 000000000..70d282e23 --- /dev/null +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelConsumerException.java @@ -0,0 +1,15 @@ +package io.confluent.parallelconsumer; + +/*- + * Copyright (C) 2020-2022 Confluent, Inc. + */ + + +/** + * Generic Parallel Consumer {@link RuntimeException} parent. + */ +public class ParallelConsumerException extends RuntimeException { + public ParallelConsumerException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractOffsetCommitter.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractOffsetCommitter.java index 1b6f9b2ae..762e7ce7e 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractOffsetCommitter.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractOffsetCommitter.java @@ -3,6 +3,7 @@ /*- * Copyright (C) 2020-2022 Confluent, Inc. */ + import io.confluent.parallelconsumer.state.WorkManager; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -25,10 +26,9 @@ public abstract class AbstractOffsetCommitter implements OffsetCommitter { @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 offsetsToCommit = wm.findCompletedEligibleOffsetsAndRemove(); + var offsetsToCommit = wm.collectCommitDataForDirtyPartitions(); if (offsetsToCommit.isEmpty()) { log.debug("No offsets ready"); } else { diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ExternalEngine.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ExternalEngine.java index 543899e31..1c863de5f 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ExternalEngine.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ExternalEngine.java @@ -4,7 +4,6 @@ * Copyright (C) 2020-2022 Confluent, Inc. */ -import com.google.common.flogger.FluentLogger; import io.confluent.parallelconsumer.ParallelConsumerOptions; import io.confluent.parallelconsumer.state.WorkContainer; import lombok.extern.slf4j.Slf4j; @@ -18,8 +17,6 @@ @Slf4j public abstract class ExternalEngine extends AbstractParallelEoSStreamProcessor { - private static final FluentLogger flog = FluentLogger.forEnclosingClass(); - protected ExternalEngine(final ParallelConsumerOptions newOptions) { super(newOptions); } diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ParallelConsumerInternalException.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ParallelConsumerInternalException.java new file mode 100644 index 000000000..96dcf471d --- /dev/null +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ParallelConsumerInternalException.java @@ -0,0 +1,14 @@ +package io.confluent.parallelconsumer.internal; + +/*- + * Copyright (C) 2020-2022 Confluent, Inc. + */ + +/** + * Generic Parallel Consumer parent exception. + */ +public class ParallelConsumerInternalException extends Exception { + public ParallelConsumerInternalException(String msg) { + super(msg); + } +} diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ProducerManager.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ProducerManager.java index 1f19ced26..e6db737f7 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ProducerManager.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ProducerManager.java @@ -1,7 +1,7 @@ package io.confluent.parallelconsumer.internal; /*- - * Copyright (C) 2020-2021 Confluent, Inc. + * Copyright (C) 2020-2022 Confluent, Inc. */ import io.confluent.csid.utils.TimeUtils; @@ -21,7 +21,6 @@ import java.lang.reflect.Field; import java.lang.reflect.Method; import java.time.Duration; -import java.time.temporal.TemporalUnit; import java.util.ConcurrentModificationException; import java.util.Map; import java.util.concurrent.Future; diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/RateLimiter.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/RateLimiter.java index d31120dd1..cafbb68e8 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/RateLimiter.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/RateLimiter.java @@ -1,11 +1,10 @@ package io.confluent.parallelconsumer.internal; /*- - * Copyright (C) 2020-2021 Confluent, Inc. + * Copyright (C) 2020-2022 Confluent, Inc. */ import lombok.Getter; -import lombok.SneakyThrows; import java.time.Duration; @@ -22,7 +21,6 @@ public RateLimiter(int seconds) { this.rate = Duration.ofSeconds(seconds); } - @SneakyThrows public void performIfNotLimited(final Runnable action) { if (isOkToCallAction()) { lastFireMs = System.currentTimeMillis(); diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/EncodingNotSupportedException.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/EncodingNotSupportedException.java index c17edd3d0..334ae159e 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/EncodingNotSupportedException.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/EncodingNotSupportedException.java @@ -1,9 +1,15 @@ package io.confluent.parallelconsumer.offsets; +/*- + * Copyright (C) 2020-2022 Confluent, Inc. + */ + +import io.confluent.parallelconsumer.internal.ParallelConsumerInternalException; + /*- * Copyright (C) 2020-2021 Confluent, Inc. */ -public class EncodingNotSupportedException extends Exception { +public class EncodingNotSupportedException extends ParallelConsumerInternalException { public EncodingNotSupportedException(final String message) { super(message); } diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/NoEncodingPossibleException.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/NoEncodingPossibleException.java new file mode 100644 index 000000000..edea86d10 --- /dev/null +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/NoEncodingPossibleException.java @@ -0,0 +1,14 @@ +package io.confluent.parallelconsumer.offsets; + +/*- + * Copyright (C) 2020-2022 Confluent, Inc. + */ + +import io.confluent.parallelconsumer.internal.ParallelConsumerInternalException; + +public class NoEncodingPossibleException extends ParallelConsumerInternalException { + + public NoEncodingPossibleException(String msg) { + super(msg); + } +} diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetEncoder.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetEncoder.java index de7f191d4..86542e0f6 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetEncoder.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetEncoder.java @@ -1,7 +1,7 @@ package io.confluent.parallelconsumer.offsets; /*- - * Copyright (C) 2020-2021 Confluent, Inc. + * Copyright (C) 2020-2022 Confluent, Inc. */ import lombok.SneakyThrows; @@ -18,7 +18,7 @@ abstract class OffsetEncoder { private final OffsetSimultaneousEncoder offsetSimultaneousEncoder; - public OffsetEncoder(OffsetSimultaneousEncoder offsetSimultaneousEncoder) { + protected OffsetEncoder(OffsetSimultaneousEncoder offsetSimultaneousEncoder) { this.offsetSimultaneousEncoder = offsetSimultaneousEncoder; } @@ -50,7 +50,8 @@ 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))); + EncodedOffsetPair encodedPair = new EncodedOffsetPair(type, ByteBuffer.wrap(bytes)); + offsetSimultaneousEncoder.sortedEncodings.add(encodedPair); offsetSimultaneousEncoder.encodingMap.put(type, bytes); } diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetMapCodecManager.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetMapCodecManager.java index b620fff67..9054b411b 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetMapCodecManager.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetMapCodecManager.java @@ -3,6 +3,7 @@ /*- * Copyright (C) 2020-2022 Confluent, Inc. */ + import io.confluent.parallelconsumer.internal.InternalRuntimeError; import io.confluent.parallelconsumer.state.PartitionState; import lombok.Value; @@ -15,7 +16,6 @@ import java.nio.charset.Charset; 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; @@ -55,7 +55,7 @@ public class OffsetMapCodecManager { public static final Charset CHARSET_TO_USE = UTF_8; - // todo OffsetMapCodecManager needs refactoring - consumer presence here smells bad + // todo OffsetMapCodecManager needs refactoring - consumer presence here smells bad #233 org.apache.kafka.clients.consumer.Consumer consumer; /** @@ -67,23 +67,23 @@ public static class HighestOffsetAndIncompletes { /** * The highest represented offset in this result. */ - Long highestSeenOffset; + Optional highestSeenOffset; /** * Of the offsets encoded, the incomplete ones. */ Set incompleteOffsets; - public static HighestOffsetAndIncompletes of(Long highestSeenOffset) { - return new HighestOffsetAndIncompletes(highestSeenOffset, new HashSet<>()); + public static HighestOffsetAndIncompletes of(long highestSeenOffset) { + return new HighestOffsetAndIncompletes(Optional.of(highestSeenOffset), new HashSet<>()); } public static HighestOffsetAndIncompletes of(long highestSeenOffset, Set incompleteOffsets) { - return new HighestOffsetAndIncompletes(highestSeenOffset, incompleteOffsets); + return new HighestOffsetAndIncompletes(Optional.of(highestSeenOffset), incompleteOffsets); } public static HighestOffsetAndIncompletes of() { - return HighestOffsetAndIncompletes.of(null); + return new HighestOffsetAndIncompletes(Optional.empty(), new HashSet<>()); } } @@ -92,26 +92,26 @@ public static HighestOffsetAndIncompletes of() { */ public static Optional forcedCodec = Optional.empty(); + // todo remove consumer #233 public OffsetMapCodecManager(final org.apache.kafka.clients.consumer.Consumer consumer) { this.consumer = consumer; } /** * Load all the previously completed offsets that were not committed - * - * @return */ + // todo this is the only method that needs the consumer - offset encoding is being conflated with decoding upon assignment #233 // todo make package private? // todo rename - public Map> loadOffsetMapForPartition(final Set assignment) { + public Map> loadPartitionStateForAssignment(final Collection assignment) { // load last committed state / metadata from consumer // todo this should be controlled for - improve consumer management so that this can't happen - Map lastCommittedOffsets = null; + Map partitionLastCommittedOffsets = null; int attempts = 0; - while (lastCommittedOffsets == null) { + while (partitionLastCommittedOffsets == null) { WakeupException lastWakeupException = null; try { - lastCommittedOffsets = consumer.committed(assignment); + partitionLastCommittedOffsets = consumer.committed(new HashSet<>(assignment)); } catch (WakeupException exception) { log.debug("Woken up trying to get assignment", exception); lastWakeupException = exception; @@ -121,15 +121,12 @@ public Map> loadOffsetMapForPartition(final throw new InternalRuntimeError("Failed to get partition assignment - continuously woken up.", lastWakeupException); } - var states = new HashMap>(); - lastCommittedOffsets.forEach((tp, offsetAndMeta) -> { + var partitionStates = new HashMap>(); + partitionLastCommittedOffsets.forEach((tp, offsetAndMeta) -> { if (offsetAndMeta != null) { - long nextExpectedOffset = offsetAndMeta.offset(); - String metadata = offsetAndMeta.metadata(); try { - // todo rename - PartitionState incompletes = decodeIncompletes(nextExpectedOffset, tp, metadata); - states.put(tp, incompletes); + PartitionState state = decodePartitionState(tp, offsetAndMeta); + partitionStates.put(tp, state); } catch (OffsetDecodingError offsetDecodingError) { log.error("Error decoding offsets from assigned partition, dropping offset map (will replay previously completed messages - partition: {}, data: {})", tp, offsetAndMeta, offsetDecodingError); @@ -138,13 +135,20 @@ public Map> loadOffsetMapForPartition(final }); - // for each assignment which isn't now added in the states to return, enter a default entry. Catches multiple other cases. + // assigned partitions for which there has never been a commit + // for each assignment with no commit history, enter a default entry. Catches multiple other cases. assignment.stream() - .filter(x -> !states.containsKey(x)) - .forEach(x -> - states.put(x, new PartitionState<>(x, HighestOffsetAndIncompletes.of()))); + .filter(topicPartition -> !partitionStates.containsKey(topicPartition)) + .forEach(topicPartition -> { + PartitionState defaultEntry = new PartitionState<>(topicPartition, HighestOffsetAndIncompletes.of()); + partitionStates.put(topicPartition, defaultEntry); + }); - return states; + return partitionStates; + } + + private HighestOffsetAndIncompletes deserialiseIncompleteOffsetMapFromBase64(OffsetAndMetadata offsetData) throws OffsetDecodingError { + return deserialiseIncompleteOffsetMapFromBase64(offsetData.offset(), offsetData.metadata()); } public static HighestOffsetAndIncompletes deserialiseIncompleteOffsetMapFromBase64(long committedOffsetForPartition, String base64EncodedOffsetPayload) throws OffsetDecodingError { @@ -157,19 +161,18 @@ public static HighestOffsetAndIncompletes deserialiseIncompleteOffsetMapFromBase return decodeCompressedOffsets(committedOffsetForPartition, decodedBytes); } - // todo rename - PartitionState decodeIncompletes(long nextExpectedOffset, TopicPartition tp, String offsetMetadataPayload) throws OffsetDecodingError { - HighestOffsetAndIncompletes incompletes = deserialiseIncompleteOffsetMapFromBase64(nextExpectedOffset, offsetMetadataPayload); + PartitionState decodePartitionState(TopicPartition tp, OffsetAndMetadata offsetData) throws OffsetDecodingError { + HighestOffsetAndIncompletes incompletes = deserialiseIncompleteOffsetMapFromBase64(offsetData); log.debug("Loaded incomplete offsets from offset payload {}", incompletes); return new PartitionState(tp, incompletes); } - public String makeOffsetMetadataPayload(long finalOffsetForPartition, PartitionState state) throws EncodingNotSupportedException { + public String makeOffsetMetadataPayload(long finalOffsetForPartition, PartitionState state) throws NoEncodingPossibleException { String offsetMap = serialiseIncompleteOffsetMapToBase64(finalOffsetForPartition, state); return offsetMap; } - String serialiseIncompleteOffsetMapToBase64(long finalOffsetForPartition, PartitionState state) throws EncodingNotSupportedException { + String serialiseIncompleteOffsetMapToBase64(long finalOffsetForPartition, PartitionState state) throws NoEncodingPossibleException { byte[] compressedEncoding = encodeOffsetsCompressed(finalOffsetForPartition, state); String b64 = OffsetSimpleSerialisation.base64(compressedEncoding); return b64; @@ -183,11 +186,11 @@ String serialiseIncompleteOffsetMapToBase64(long finalOffsetForPartition, Partit *

* Can remove string encoding in favour of the boolean array for the `BitSet` if that's how things settle. */ - byte[] encodeOffsetsCompressed(long finalOffsetForPartition, PartitionState partition) throws EncodingNotSupportedException { - Set incompleteOffsets = partition.getIncompleteOffsets(); - log.debug("Encoding partition {} incomplete offsets {}", partition.getTp(), incompleteOffsets); - long offsetHighestSucceeded = partition.getOffsetHighestSucceeded(); - OffsetSimultaneousEncoder simultaneousEncoder = new OffsetSimultaneousEncoder(finalOffsetForPartition, offsetHighestSucceeded, incompleteOffsets).invoke(); + byte[] encodeOffsetsCompressed(long finalOffsetForPartition, PartitionState partitionState) throws NoEncodingPossibleException { + var incompleteOffsets = partitionState.getIncompleteOffsetsBelowHighestSucceeded(); + long highestSucceeded = partitionState.getOffsetHighestSucceeded(); + log.debug("Encoding partition {}: highest suceeded {}, incomplete offsets {}, ", partitionState.getTp(), highestSucceeded, incompleteOffsets); + OffsetSimultaneousEncoder simultaneousEncoder = new OffsetSimultaneousEncoder(finalOffsetForPartition, highestSucceeded, incompleteOffsets).invoke(); // if (forcedCodec.isPresent()) { @@ -196,7 +199,7 @@ byte[] encodeOffsetsCompressed(long finalOffsetForPartition, PartitionState 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)); + throw new NoEncodingPossibleException(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(); @@ -219,48 +222,8 @@ static HighestOffsetAndIncompletes decodeCompressedOffsets(long nextExpectedOffs return HighestOffsetAndIncompletes.of(highestSeenOffsetIsThen); } else { var result = EncodedOffsetPair.unwrap(decodedBytes); - - HighestOffsetAndIncompletes incompletesTuple = result.getDecodedIncompletes(nextExpectedOffset); - - Set incompletes = incompletesTuple.getIncompleteOffsets(); - long highWater = incompletesTuple.getHighestSeenOffset(); - - return HighestOffsetAndIncompletes.of(highWater, incompletes); + return result.getDecodedIncompletes(nextExpectedOffset); } } - String incompletesToBitmapString(long finalOffsetForPartition, PartitionState state) { - var runLengthString = new StringBuilder(); - Long lowWaterMark = finalOffsetForPartition; - Long highWaterMark = state.getOffsetHighestSeen(); - long end = highWaterMark - lowWaterMark; - for (final var relativeOffset : range(end)) { - long offset = lowWaterMark + relativeOffset; - if (state.getIncompleteOffsets().contains(offset)) { - runLengthString.append("o"); - } else { - runLengthString.append("x"); - } - } - return runLengthString.toString(); - } - - static Set bitmapStringToIncomplete(final long baseOffset, final String inputBitmapString) { - final Set incompleteOffsets = new HashSet<>(); - - final long longLength = inputBitmapString.length(); - range(longLength).forEach(i -> { - var bit = inputBitmapString.charAt(i); - if (bit == 'o') { - incompleteOffsets.add(baseOffset + i); - } else if (bit == 'x') { - log.trace("Dropping completed offset"); - } else { - throw new IllegalArgumentException("Invalid encoding - unexpected char: " + bit); - } - }); - - return incompleteOffsets; - } - } diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetSimultaneousEncoder.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetSimultaneousEncoder.java index e9479a867..237bccdad 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetSimultaneousEncoder.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetSimultaneousEncoder.java @@ -1,7 +1,7 @@ package io.confluent.parallelconsumer.offsets; /*- - * Copyright (C) 2020-2021 Confluent, Inc. + * Copyright (C) 2020-2022 Confluent, Inc. */ import io.confluent.parallelconsumer.state.WorkManager; @@ -56,12 +56,12 @@ public class OffsetSimultaneousEncoder { Map encodingMap = new EnumMap<>(OffsetEncoding.class); /** - * Ordered set of the the different encodings, used to quickly retrieve the most compressed encoding + * Ordered set of the different encodings, used to quickly retrieve the most compressed encoding * * @see #packSmallest() */ @Getter - PriorityQueue sortedEncodings = new PriorityQueue(); + PriorityQueue sortedEncodings = new PriorityQueue<>(); /** @@ -80,7 +80,7 @@ public class OffsetSimultaneousEncoder { /** * The encoders to run */ - private final Set encoders = new HashSet<>(); + private final Set encoders; public OffsetSimultaneousEncoder(long lowWaterMark, long highestSucceededOffset, Set incompleteOffsets) { this.lowWaterMark = lowWaterMark; @@ -101,28 +101,31 @@ public OffsetSimultaneousEncoder(long lowWaterMark, long highestSucceededOffset, // sanity if (bitsetLengthL != length) throw new IllegalArgumentException("Integer overflow"); - initEncoders(); + this.encoders = initEncoders(); } - private void initEncoders() { + private Set initEncoders() { + var newEncoders = new HashSet(); if (length > LARGE_INPUT_MAP_SIZE_THRESHOLD) { log.debug("~Large input map size: {} (start: {} end: {})", length, lowWaterMark, lowWaterMark + length); } try { - encoders.add(new BitSetEncoder(length, this, v1)); + newEncoders.add(new BitSetEncoder(length, this, v1)); } catch (BitSetEncodingNotSupportedException a) { log.debug("Cannot use {} encoder ({})", BitSetEncoder.class.getSimpleName(), a.getMessage()); } try { - encoders.add(new BitSetEncoder(length, this, v2)); + newEncoders.add(new BitSetEncoder(length, this, v2)); } catch (BitSetEncodingNotSupportedException a) { log.warn("Cannot use {} encoder ({})", BitSetEncoder.class.getSimpleName(), a.getMessage()); } - encoders.add(new RunLengthEncoder(this, v1)); - encoders.add(new RunLengthEncoder(this, v2)); + newEncoders.add(new RunLengthEncoder(this, v1)); + newEncoders.add(new RunLengthEncoder(this, v2)); + + return newEncoders; } /** @@ -204,7 +207,7 @@ private void registerEncodings(final Set encoders) { toRemove.add(encoder); } } - encoders.removeAll(toRemove); + toRemove.forEach(encoders::remove); // compressed versions // sizes over LARGE_INPUT_MAP_SIZE_THRESHOLD bytes seem to benefit from compression @@ -219,9 +222,9 @@ private void registerEncodings(final Set encoders) { * * @see #packEncoding(EncodedOffsetPair) */ - public byte[] packSmallest() throws EncodingNotSupportedException { + public byte[] packSmallest() throws NoEncodingPossibleException { if (sortedEncodings.isEmpty()) { - throw new EncodingNotSupportedException("No encodings could be used"); + throw new NoEncodingPossibleException("No encodings could be used"); } final EncodedOffsetPair best = this.sortedEncodings.poll(); log.debug("Compression chosen is: {}", best.encoding.name()); diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionMonitor.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionMonitor.java index 4e7a08d87..65c19b2dd 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionMonitor.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionMonitor.java @@ -9,7 +9,6 @@ import io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor; import io.confluent.parallelconsumer.internal.BrokerPollSystem; import io.confluent.parallelconsumer.internal.InternalRuntimeError; -import io.confluent.parallelconsumer.offsets.EncodingNotSupportedException; import io.confluent.parallelconsumer.offsets.OffsetMapCodecManager; import lombok.Getter; import lombok.RequiredArgsConstructor; @@ -20,17 +19,17 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.common.TopicPartition; -import pl.tlinkowski.unij.api.UniSets; import java.time.Clock; -import java.util.*; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; import static io.confluent.csid.utils.KafkaUtils.toTopicPartition; import static io.confluent.csid.utils.StringUtils.msg; -import static io.confluent.parallelconsumer.offsets.OffsetMapCodecManager.DefaultMaxMetadataSize; /** * In charge of managing {@link PartitionState}s. @@ -49,13 +48,13 @@ public class PartitionMonitor implements ConsumerRebalanceListener { */ @Getter @Setter - private double USED_PAYLOAD_THRESHOLD_MULTIPLIER = 0.75; + private static double USED_PAYLOAD_THRESHOLD_MULTIPLIER = 0.75; private final Consumer consumer; private final ShardManager sm; - private final ParallelConsumerOptions options; + private final ParallelConsumerOptions options; /** * Hold the tracking state for each of our managed partitions. @@ -71,15 +70,7 @@ public class PartitionMonitor implements ConsumerRebalanceListener { */ private final Map partitionsAssignmentEpochs = new ConcurrentHashMap<>(); - /** - * Gets set to true whenever work is returned completed, so that we know when a commit needs to be made. - *

- * In normal operation, this probably makes very little difference, as typical commit frequency is 1 second, so low - * chances no work has completed in the last second. - */ - private final AtomicBoolean workStateIsDirtyNeedsCommitting = new AtomicBoolean(false); - - final private Clock clock; + private final Clock clock; public PartitionState getPartitionState(TopicPartition tp) { // may cause the system to wait for a rebalance to finish @@ -90,32 +81,32 @@ public PartitionState getPartitionState(TopicPartition tp) { } /** - * Load offset map for assigned partitions + * Load offset map for assigned assignedPartitions */ @Override - public void onPartitionsAssigned(Collection partitions) { - log.debug("Partitions assigned: {}", partitions); + public void onPartitionsAssigned(Collection assignedPartitions) { + log.debug("Partitions assigned: {}", assignedPartitions); synchronized (this.partitionStates) { - for (final TopicPartition partition : partitions) { - if (this.partitionStates.containsKey(partition)) { - PartitionState state = partitionStates.get(partition); - if (state.isRemoved()) { - log.trace("Reassignment of previously revoked partition {} - state: {}", partition, state); + for (final TopicPartition partitionAssignment : assignedPartitions) { + boolean isAlreadyAssigned = this.partitionStates.containsKey(partitionAssignment); + if (isAlreadyAssigned) { + PartitionState previouslyAssignedState = partitionStates.get(partitionAssignment); + if (previouslyAssignedState.isRemoved()) { + log.trace("Reassignment of previously revoked partition {} - state: {}", partitionAssignment, previouslyAssignedState); } else { log.warn("New assignment of partition which already exists and isn't recorded as removed in " + "partition state. Could be a state bug - was the partition revocation somehow missed, " + - "or is this a race? Please file a GH issue. Partition: {}, state: {}", partition, state); + "or is this a race? Please file a GH issue. Partition: {}, state: {}", partitionAssignment, previouslyAssignedState); } } } - incrementPartitionAssignmentEpoch(partitions); + incrementPartitionAssignmentEpoch(assignedPartitions); try { - Set partitionsSet = UniSets.copyOf(partitions); - OffsetMapCodecManager om = new OffsetMapCodecManager<>(this.consumer); // todo remove throw away instance creation - var partitionStates = om.loadOffsetMapForPartition(partitionsSet); + OffsetMapCodecManager om = new OffsetMapCodecManager<>(this.consumer); // todo remove throw away instance creation - #233 + var partitionStates = om.loadPartitionStateForAssignment(assignedPartitions); this.partitionStates.putAll(partitionStates); } catch (Exception e) { log.error("Error in onPartitionsAssigned", e); @@ -203,8 +194,7 @@ private void resetOffsetMapAndRemoveWork(Collection allRemovedPa partitionStates.put(removedPartition, RemovedPartitionState.getSingleton()); // - NavigableMap> workFromRemovedPartition = partition.getCommitQueue(); - sm.removeAnyShardsReferencedBy(workFromRemovedPartition); + partition.onPartitionsRemoved(sm); } } @@ -252,12 +242,7 @@ boolean checkIfWorkIsStale(final WorkContainer workContainer) { return false; } - public void maybeRaiseHighestSeenOffset(TopicPartition tp, long seenOffset) { - PartitionState partitionState = getPartitionState(tp); - partitionState.maybeRaiseHighestSeenOffset(seenOffset); - } - - boolean isRecordPreviouslyCompleted(ConsumerRecord rec) { + public boolean isRecordPreviouslyCompleted(ConsumerRecord rec) { var tp = toTopicPartition(rec); var partitionState = getPartitionState(tp); boolean previouslyCompleted = partitionState.isRecordPreviouslyCompleted(rec); @@ -296,12 +281,7 @@ public long getNumberOfEntriesInPartitionQueues() { .orElse(0); } - private void setPartitionMoreRecordsAllowedToProcess(TopicPartition topicPartitionKey, boolean moreMessagesAllowed) { - var state = getPartitionState(topicPartitionKey); - state.setAllowedMoreRecords(moreMessagesAllowed); - } - - public Long getHighestSeenOffset(final TopicPartition tp) { + public long getHighestSeenOffset(final TopicPartition tp) { return getPartitionState(tp).getOffsetHighestSeen(); } @@ -327,75 +307,6 @@ public boolean isBlocked(final TopicPartition topicPartition) { return !isAllowedMoreRecords(topicPartition); } - /** - * Get final offset data, build the offset map, and replace it in our map of offset data to send - * - * @param offsetsToSend - * @param topicPartitionKey - * @param incompleteOffsets - */ - //todo refactor - void addEncodedOffsets(Map offsetsToSend, - TopicPartition topicPartitionKey, - LinkedHashSet incompleteOffsets) { - // TODO potential optimisation: store/compare the current incomplete offsets to the last committed ones, to know if this step is needed or not (new progress has been made) - isdirty? - boolean incompleteOffsetsNeedingEncoding = !incompleteOffsets.isEmpty(); - if (incompleteOffsetsNeedingEncoding) { - // todo offsetOfNextExpectedMessage should be an attribute of State - consider deriving it from the state class - long offsetOfNextExpectedMessage; - OffsetAndMetadata finalOffsetOnly = offsetsToSend.get(topicPartitionKey); - if (finalOffsetOnly == null) { - // no new low watermark to commit, so use the last one again - offsetOfNextExpectedMessage = incompleteOffsets.iterator().next(); // first element - } else { - offsetOfNextExpectedMessage = finalOffsetOnly.offset(); - } - - OffsetMapCodecManager om = new OffsetMapCodecManager<>(this.consumer); - try { - PartitionState state = getPartitionState(topicPartitionKey); - // todo smelly - update the partition state with the new found incomplete offsets. This field is used by nested classes accessing the state - state.setIncompleteOffsets(incompleteOffsets); - String offsetMapPayload = om.makeOffsetMetadataPayload(offsetOfNextExpectedMessage, state); - int metaPayloadLength = offsetMapPayload.length(); - boolean moreMessagesAllowed; - OffsetAndMetadata offsetWithExtraMap; - // todo move - double pressureThresholdValue = DefaultMaxMetadataSize * USED_PAYLOAD_THRESHOLD_MULTIPLIER; - - if (metaPayloadLength > DefaultMaxMetadataSize) { - // exceeded maximum API allowed, strip the payload - moreMessagesAllowed = false; - offsetWithExtraMap = new OffsetAndMetadata(offsetOfNextExpectedMessage); // strip payload - log.warn("Offset map data too large (size: {}) to fit in metadata payload hard limit of {} - cannot include in commit. " + - "Warning: messages might be replayed on rebalance. " + - "See kafka.coordinator.group.OffsetConfig#DefaultMaxMetadataSize = {} and issue #47.", - metaPayloadLength, DefaultMaxMetadataSize, DefaultMaxMetadataSize); - } else if (metaPayloadLength > pressureThresholdValue) { // and thus metaPayloadLength <= DefaultMaxMetadataSize - // try to turn on back pressure before max size is reached - moreMessagesAllowed = false; - offsetWithExtraMap = new OffsetAndMetadata(offsetOfNextExpectedMessage, offsetMapPayload); - log.warn("Payload size {} higher than threshold {}, but still lower than max {}. Will write payload, but will " + - "not allow further messages, in order to allow the offset data to shrink (via succeeding messages).", - metaPayloadLength, pressureThresholdValue, DefaultMaxMetadataSize); - } else { // and thus (metaPayloadLength <= pressureThresholdValue) - moreMessagesAllowed = true; - offsetWithExtraMap = new OffsetAndMetadata(offsetOfNextExpectedMessage, offsetMapPayload); - log.debug("Payload size {} within threshold {}", metaPayloadLength, pressureThresholdValue); - } - - setPartitionMoreRecordsAllowedToProcess(topicPartitionKey, moreMessagesAllowed); - offsetsToSend.put(topicPartitionKey, offsetWithExtraMap); - } catch (EncodingNotSupportedException e) { - setPartitionMoreRecordsAllowedToProcess(topicPartitionKey, false); - log.warn("No encodings could be used to encode the offset map, skipping. Warning: messages might be replayed on rebalance.", e); - } - } else { - setPartitionMoreRecordsAllowedToProcess(topicPartitionKey, true); - } - } - - public boolean isPartitionRemovedOrNeverAssigned(ConsumerRecord rec) { TopicPartition topicPartition = toTopicPartition(rec); var partitionState = getPartitionState(topicPartition); @@ -408,6 +319,11 @@ public void onSuccess(WorkContainer wc) { partitionState.onSuccess(wc); } + public void onFailure(WorkContainer wc) { + PartitionState partitionState = getPartitionState(wc.getTopicPartition()); + partitionState.onFailure(wc); + } + /** * Takes a record as work and puts it into internal queues, unless it's been previously recorded as completed as per * loaded records. @@ -433,8 +349,7 @@ boolean maybeRegisterNewRecordAsWork(final ConsumerRecord rec) { return false; } else { int currentPartitionEpoch = getEpoch(rec); - //noinspection unchecked - Lombok builder getter erases generics - var wc = new WorkContainer(currentPartitionEpoch, rec, options.getRetryDelayProvider(), clock); + var wc = new WorkContainer<>(currentPartitionEpoch, rec, options.getRetryDelayProvider(), clock); sm.addWorkContainer(wc); @@ -445,92 +360,13 @@ boolean maybeRegisterNewRecordAsWork(final ConsumerRecord rec) { } } - public Map findCompletedEligibleOffsetsAndRemove() { - return findCompletedEligibleOffsetsAndRemove(true); - } - - /** - * Finds eligible offset positions to commit in each assigned partition - */ - // todo remove completely as state of offsets should be tracked live, no need to scan for them - see #201 - // https://github.com/confluentinc/parallel-consumer/issues/201 Refactor: Live tracking of offsets as they change, so we don't need to scan for them - Map findCompletedEligibleOffsetsAndRemove(boolean remove) { - - // - Map offsetsToSend = new HashMap<>(); - int count = 0; - int removed = 0; - log.trace("Scanning for in order in-flight work that has completed..."); - - // - for (var partitionStateEntry : getAssignedPartitions().entrySet()) { - var partitionState = partitionStateEntry.getValue(); - Map> partitionQueue = partitionState.getCommitQueue(); - TopicPartition topicPartitionKey = partitionStateEntry.getKey(); - log.trace("Starting scan of partition: {}", topicPartitionKey); - - count += partitionQueue.size(); - var workToRemove = new LinkedList>(); - var incompleteOffsets = new LinkedHashSet(); - long lowWaterMark = -1; - var highestSucceeded = partitionState.getOffsetHighestSucceeded(); - // can't commit this offset or beyond, as this is the latest offset that is incomplete - // i.e. only commit offsets that come before the current one, and stop looking for more - boolean beyondSuccessiveSucceededOffsets = false; - for (final var offsetAndItsWorkContainer : partitionQueue.entrySet()) { - // ordered iteration via offset keys thanks to the tree-map - WorkContainer container = offsetAndItsWorkContainer.getValue(); - - // - long offset = container.getCr().offset(); - if (offset > highestSucceeded) { - break; // no more to encode - } - - // - boolean complete = container.isUserFunctionComplete(); - if (complete) { - if (container.getUserFunctionSucceeded().get() && !beyondSuccessiveSucceededOffsets) { - log.trace("Found offset candidate ({}) to add to offset commit map", container); - workToRemove.add(container); - // as in flights are processed in order, this will keep getting overwritten with the highest offset available - // current offset is the highest successful offset, so commit +1 - offset to be committed is defined as the offset of the next expected message to be read - long offsetOfNextExpectedMessageToBeCommitted = offset + 1; - OffsetAndMetadata offsetData = new OffsetAndMetadata(offsetOfNextExpectedMessageToBeCommitted); - offsetsToSend.put(topicPartitionKey, offsetData); - } else if (container.getUserFunctionSucceeded().get() && beyondSuccessiveSucceededOffsets) { - // todo lookup the low water mark and include here - log.trace("Offset {} is complete and succeeded, but we've iterated past the lowest committable offset ({}). Will mark as complete in the offset map.", - container.getCr().offset(), lowWaterMark); - // no-op - offset map is only for not succeeded or completed offsets - } else { - log.trace("Offset {} is complete, but failed processing. Will track in offset map as failed. Can't do normal offset commit past this point.", container.getCr().offset()); - beyondSuccessiveSucceededOffsets = true; - incompleteOffsets.add(offset); - } - } else { - lowWaterMark = container.offset(); - beyondSuccessiveSucceededOffsets = true; - log.trace("Offset (:{}) is incomplete, holding up the queue ({}) of size {}.", - container.getCr().offset(), - topicPartitionKey, - partitionQueue.size()); - incompleteOffsets.add(offset); - } - } - - addEncodedOffsets(offsetsToSend, topicPartitionKey, incompleteOffsets); - - if (remove) { - removed += workToRemove.size(); - partitionState.remove(workToRemove); - } + public Map collectDirtyCommitData() { + var dirties = new HashMap(); + for (var state : getAssignedPartitions().values()) { + var offsetAndMetadata = state.getCommitDataIfDirty(); + offsetAndMetadata.ifPresent(andMetadata -> dirties.put(state.getTp(), andMetadata)); } - - - log.debug("Scan finished, {} were in flight, {} completed offsets removed, coalesced to {} offset(s) ({}) to be committed", - count, removed, offsetsToSend.size(), offsetsToSend); - return offsetsToSend; + return dirties; } private Map> getAssignedPartitions() { diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionState.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionState.java index ffed07e9a..e2ccf4094 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionState.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionState.java @@ -4,9 +4,10 @@ * Copyright (C) 2020-2022 Confluent, Inc. */ +import io.confluent.parallelconsumer.internal.BrokerPollSystem; +import io.confluent.parallelconsumer.offsets.NoEncodingPossibleException; import io.confluent.parallelconsumer.offsets.OffsetMapCodecManager; import lombok.Getter; -import lombok.NonNull; import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -14,11 +15,16 @@ import org.apache.kafka.common.TopicPartition; import java.util.Collections; -import java.util.LinkedList; import java.util.NavigableMap; +import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.ConcurrentSkipListSet; +import java.util.stream.Collectors; +import static io.confluent.parallelconsumer.offsets.OffsetMapCodecManager.DefaultMaxMetadataSize; +import static java.util.Optional.empty; +import static java.util.Optional.of; import static lombok.AccessLevel.*; @Slf4j @@ -28,46 +34,43 @@ public class PartitionState { private final TopicPartition tp; /** - * A subset of Offsets, beyond the highest committable offset, which haven't been totally completed. + * Offset data beyond the highest committable offset, which haven't totally succeeded. *

- * We only need to know the full incompletes when we do the {@link #findCompletedEligibleOffsetsAndRemove} scan, so - * find the full sent only then, and discard. Otherwise, for continuous encoding, the encoders track it them - * selves. + * This is independent of the actual queued {@link WorkContainer}s. This is because to start with, data about + * incomplete offsets come from the encoded metadata payload that gets committed along with the highest committable + * offset ({@link #getOffsetHighestSequentialSucceeded()}). They are not always in sync. *

- * We work with incompletes, instead of completes, because it's a bet that most of the time the storage space for - * storing the incompletes in memory will be smaller. - * - * @see #findCompletedEligibleOffsetsAndRemove(boolean) - * @see #encodeWorkResult(boolean, WorkContainer) - * @see #onSuccess(WorkContainer) - * @see #onFailure(WorkContainer) + * TreeSet so we can always get the lowest offset. + *

+ * Needs to be concurrent because, the committer requesting the data to commit may be another thread - the broker + * polling sub system - {@link BrokerPollSystem#maybeDoCommit}. The alternative to having this as a concurrent + * collection, would be to have the control thread prepare possible commit data on every cycle, and park that data + * so that the broker polling thread can grab it, if it wants to commit - i.e. the poller would not prepare/query + * the data for itself. See also #200 Refactor: Consider a shared nothing architecture. */ - // visible for testing - // todo should be tracked live, as we know when the state of work containers flips - i.e. they are continuously tracked - // this is derived from partitionCommitQueues WorkContainer states - // todo remove setter - leaky abstraction, shouldn't be needed - @Setter - private Set incompleteOffsets; + private final ConcurrentSkipListSet incompleteOffsets; - public Set getIncompleteOffsets() { - return Collections.unmodifiableSet(incompleteOffsets); - } + /** + * Cache view of the state of the partition. Is set dirty when the incomplete state of any offset changes. Is set + * clean after a successful commit of the state. + */ + @Setter(PRIVATE) + @Getter(PRIVATE) + private boolean dirty; /** * The highest seen offset for a partition. *

- * Starts off as null - no data + * Starts off as -1 - no data. Offsets in Kafka are never negative, so this is fine. */ // visible for testing - @NonNull @Getter(PUBLIC) - private Long offsetHighestSeen; + private long offsetHighestSeen; /** * Highest offset which has completed successfully ("succeeded"). */ @Getter(PUBLIC) - @Setter(PRIVATE) private long offsetHighestSucceeded = -1L; /** @@ -84,7 +87,7 @@ public Set getIncompleteOffsets() { * @see OffsetMapCodecManager#DefaultMaxMetadataSize */ @Getter(PACKAGE) - @Setter(PACKAGE) + @Setter(PRIVATE) private boolean allowedMoreRecords = true; /** @@ -94,72 +97,77 @@ public Set getIncompleteOffsets() { * advancing offsets, as this isn't a guarantee of kafka's. *

* Concurrent because either the broker poller thread or the control thread may be requesting offset to commit - * ({@link #findCompletedEligibleOffsetsAndRemove}) - * - * @see #findCompletedEligibleOffsetsAndRemove + * ({@link #getCommitDataIfDirty()}), or reading upon {@link #onPartitionsRemoved} */ + // todo doesn't need to be concurrent any more? private final NavigableMap> commitQueue = new ConcurrentSkipListMap<>(); - NavigableMap> getCommitQueue() { + private NavigableMap> getCommitQueue() { return Collections.unmodifiableNavigableMap(commitQueue); } - public PartitionState(TopicPartition tp, OffsetMapCodecManager.HighestOffsetAndIncompletes incompletes) { + public PartitionState(TopicPartition tp, OffsetMapCodecManager.HighestOffsetAndIncompletes offsetData) { this.tp = tp; - this.incompleteOffsets = incompletes.getIncompleteOffsets(); - this.offsetHighestSeen = incompletes.getHighestSeenOffset(); + this.offsetHighestSeen = offsetData.getHighestSeenOffset().orElse(-1L); + this.incompleteOffsets = new ConcurrentSkipListSet<>(offsetData.getIncompleteOffsets()); + this.offsetHighestSucceeded = this.offsetHighestSeen; } - public void maybeRaiseHighestSeenOffset(final long highestSeen) { - // rise the high water mark - Long oldHighestSeen = this.offsetHighestSeen; - if (oldHighestSeen == null || highestSeen >= oldHighestSeen) { - log.trace("Updating highest seen - was: {} now: {}", offsetHighestSeen, highestSeen); - offsetHighestSeen = highestSeen; + private void maybeRaiseHighestSeenOffset(final long offset) { + // rise the highest seen offset + if (offset >= offsetHighestSeen) { + log.trace("Updating highest seen - was: {} now: {}", offsetHighestSeen, offset); + offsetHighestSeen = offset; } } - /** - * Removes all offsets that fall below the new low water mark. - */ - public void truncateOffsets(final long nextExpectedOffset) { - incompleteOffsets.removeIf(offset -> offset < nextExpectedOffset); + public void onOffsetCommitSuccess(final OffsetAndMetadata committed) { + setClean(); } - public void onOffsetCommitSuccess(final OffsetAndMetadata committed) { - long nextExpectedOffset = committed.offset(); - truncateOffsets(nextExpectedOffset); + private void setClean() { + setDirty(false); + } + + private void setDirty() { + setDirty(true); } public boolean isRecordPreviouslyCompleted(final ConsumerRecord rec) { - boolean previouslyProcessed; - long offset = rec.offset(); - if (incompleteOffsets.contains(offset)) { - // record previously saved as not being completed, can exit early - previouslyProcessed = false; + long recOffset = rec.offset(); + if (!incompleteOffsets.contains(recOffset)) { + // if within the range of tracked offsets, must have been previously completed, as it's not in the incomplete set + return recOffset <= offsetHighestSeen; } else { - Long offsetHighWaterMark = offsetHighestSeen; - // within the range of tracked offsets, so must have been previously completed // we haven't recorded this far up, so must not have been processed yet - previouslyProcessed = offsetHighWaterMark != null && offset <= offsetHighWaterMark; + return false; } - return previouslyProcessed; } public boolean hasWorkInCommitQueue() { return !commitQueue.isEmpty(); } - public boolean hasWorkThatNeedsCommitting() { - return commitQueue.values().parallelStream().anyMatch(x -> x.isUserFunctionSucceeded()); - } - public int getCommitQueueSize() { return commitQueue.size(); } public void onSuccess(WorkContainer work) { + long offset = work.offset(); + + WorkContainer removedFromQueue = this.commitQueue.remove(work.offset()); + assert (removedFromQueue != null); + + boolean removedFromIncompletes = this.incompleteOffsets.remove(offset); + assert (removedFromIncompletes); + updateHighestSucceededOffsetSoFar(work); + + setDirty(); + } + + public void onFailure(WorkContainer work) { + // no-op } /** @@ -170,21 +178,14 @@ private void updateHighestSucceededOffsetSoFar(WorkContainer work) { long thisOffset = work.offset(); if (thisOffset > highestSucceeded) { log.trace("Updating highest completed - was: {} now: {}", highestSucceeded, thisOffset); - setOffsetHighestSucceeded(thisOffset); + this.offsetHighestSucceeded = thisOffset; } } public void addWorkContainer(WorkContainer wc) { maybeRaiseHighestSeenOffset(wc.offset()); - NavigableMap> queue = this.commitQueue; - queue.put(wc.offset(), wc); - } - - public void remove(LinkedList> workToRemove) { - for (var workContainer : workToRemove) { - var offset = workContainer.getCr().offset(); - this.commitQueue.remove(offset); - } + commitQueue.put(wc.offset(), wc); + incompleteOffsets.add(wc.offset()); } /** @@ -196,4 +197,119 @@ public boolean isRemoved() { return false; } + public Optional getCommitDataIfDirty() { + if (isDirty()) + return of(createOffsetAndMetadata()); + else + return empty(); + } + + private OffsetAndMetadata createOffsetAndMetadata() { + Optional payloadOpt = tryToEncodeOffsets(); + long nextOffset = getNextExpectedPolledOffset(); + return payloadOpt + .map(s -> new OffsetAndMetadata(nextOffset, s)) + .orElseGet(() -> new OffsetAndMetadata(nextOffset)); + } + + private long getNextExpectedPolledOffset() { + return getOffsetHighestSequentialSucceeded() + 1; + } + + /** + * @return all incomplete offsets of buffered work in this shard, even if higher than the highest succeeded + */ + public Set getAllIncompleteOffsets() { + //noinspection FuseStreamOperations - only in java 10 + return Collections.unmodifiableSet(incompleteOffsets.parallelStream() + .collect(Collectors.toSet())); + } + + /** + * @return incomplete offsets which are lower than the highest succeeded + */ + public Set getIncompleteOffsetsBelowHighestSucceeded() { + long highestSucceeded = getOffsetHighestSucceeded(); + return Collections.unmodifiableSet(incompleteOffsets.parallelStream() + // todo less than or less than and equal? + .filter(x -> x < highestSucceeded) + .collect(Collectors.toSet())); + } + + public long getOffsetHighestSequentialSucceeded() { + if (this.incompleteOffsets.isEmpty()) { + return this.offsetHighestSeen; + } else { + return this.incompleteOffsets.first() - 1; + } + } + + /** + * Tries to encode the incomplete offsets for this partition. This may not be possible if there are none, or if no + * encodings are possible ({@link NoEncodingPossibleException}. Encoding may not be possible of - see {@link + * OffsetMapCodecManager#makeOffsetMetadataPayload}. + * + * @return if possible, the String encoded offset map + */ + private Optional tryToEncodeOffsets() { + if (incompleteOffsets.isEmpty()) { + setAllowedMoreRecords(true); + return empty(); + } + + try { + // todo refactor use of null shouldn't be needed. Is OffsetMapCodecManager stateful? remove null #233 + OffsetMapCodecManager om = new OffsetMapCodecManager<>(null); + long offsetOfNextExpectedMessage = getNextExpectedPolledOffset(); + String offsetMapPayload = om.makeOffsetMetadataPayload(offsetOfNextExpectedMessage, this); + boolean mustStrip = updateBlockFromEncodingResult(offsetMapPayload); + if (mustStrip) { + return empty(); + } else { + return of(offsetMapPayload); + } + } catch (NoEncodingPossibleException e) { + setAllowedMoreRecords(false); + log.warn("No encodings could be used to encode the offset map, skipping. Warning: messages might be replayed on rebalance.", e); + return empty(); + } + } + + /** + * @return true if the payload is too large and must be stripped + */ + private boolean updateBlockFromEncodingResult(String offsetMapPayload) { + int metaPayloadLength = offsetMapPayload.length(); + boolean mustStrip = false; + + if (metaPayloadLength > DefaultMaxMetadataSize) { + // exceeded maximum API allowed, strip the payload + mustStrip = true; + setAllowedMoreRecords(false); + log.warn("Offset map data too large (size: {}) to fit in metadata payload hard limit of {} - cannot include in commit. " + + "Warning: messages might be replayed on rebalance. " + + "See kafka.coordinator.group.OffsetConfig#DefaultMaxMetadataSize = {} and issue #47.", + metaPayloadLength, DefaultMaxMetadataSize, DefaultMaxMetadataSize); + } else if (metaPayloadLength > getPressureThresholdValue()) { // and thus metaPayloadLength <= DefaultMaxMetadataSize + // try to turn on back pressure before max size is reached + setAllowedMoreRecords(false); + log.warn("Payload size {} higher than threshold {}, but still lower than max {}. Will write payload, but will " + + "not allow further messages, in order to allow the offset data to shrink (via succeeding messages).", + metaPayloadLength, getPressureThresholdValue(), DefaultMaxMetadataSize); + + } else { // and thus (metaPayloadLength <= pressureThresholdValue) + setAllowedMoreRecords(true); + log.debug("Payload size {} within threshold {}", metaPayloadLength, getPressureThresholdValue()); + } + + return mustStrip; + } + + private double getPressureThresholdValue() { + return DefaultMaxMetadataSize * PartitionMonitor.getUSED_PAYLOAD_THRESHOLD_MULTIPLIER(); + } + + public void onPartitionsRemoved(ShardManager sm) { + sm.removeAnyShardsReferencedBy(getCommitQueue()); + } } diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ProcessingShard.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ProcessingShard.java index c1a6ae65a..22ff310fd 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ProcessingShard.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ProcessingShard.java @@ -12,10 +12,7 @@ import lombok.extern.slf4j.Slf4j; import java.time.Duration; -import java.util.ArrayList; -import java.util.List; -import java.util.NavigableMap; -import java.util.Optional; +import java.util.*; import java.util.concurrent.ConcurrentSkipListMap; import java.util.stream.Collectors; @@ -90,7 +87,7 @@ public WorkContainer remove(long offset) { ArrayList> getWorkIfAvailable(int workToGetDelta) { log.trace("Looking for work on shardQueueEntry: {}", getKey()); - var slowWork = new ArrayList>(); + var slowWork = new HashSet>(); var workTaken = new ArrayList>(); var iterator = entries.entrySet().iterator(); @@ -125,7 +122,7 @@ ArrayList> getWorkIfAvailable(int workToGetDelta) { return workTaken; } - private void logSlowWork(ArrayList> slowWork) { + private void logSlowWork(Set> slowWork) { // log if (!slowWork.isEmpty()) { List slowTopics = slowWork.parallelStream() @@ -137,7 +134,7 @@ private void logSlowWork(ArrayList> slowWork) { } } - private void addToSlowWorkMaybe(ArrayList> slowWork, WorkContainer workContainer) { + private void addToSlowWorkMaybe(Set> slowWork, WorkContainer workContainer) { var msgTemplate = "Can't take as work: Work ({}). Must all be true: Delay passed= {}. Is not in flight= {}. Has not succeeded already= {}. Time spent in execution queue: {}."; Duration timeInFlight = workContainer.getTimeInFlight(); var msg = msg(msgTemplate, workContainer, workContainer.hasDelayPassed(), workContainer.isNotInFlight(), !workContainer.isUserFunctionSucceeded(), timeInFlight); diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/RemovedPartitionState.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/RemovedPartitionState.java index f1928c61b..50b2f9749 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/RemovedPartitionState.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/RemovedPartitionState.java @@ -6,10 +6,8 @@ import io.confluent.csid.utils.KafkaUtils; import io.confluent.parallelconsumer.offsets.OffsetMapCodecManager; -import lombok.NonNull; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.common.TopicPartition; import java.util.*; @@ -62,14 +60,6 @@ public void addWorkContainer(final WorkContainer wc) { log.warn("Dropping new work container for partition no longer assigned. WC: {}", wc); } - @Override - public void remove(final LinkedList> workToRemove) { - if (!workToRemove.isEmpty()) { - // no-op - log.debug("Dropping work container to remove for partition no longer assigned. WC: {}", workToRemove); - } - } - /** * Don't allow more records to be processed for this partition. Eventually these records triggering this check will * be cleaned out. @@ -83,55 +73,24 @@ boolean isAllowedMoreRecords() { } @Override - public Set getIncompleteOffsets() { + public Set getIncompleteOffsetsBelowHighestSucceeded() { log.debug(NO_OP); //noinspection unchecked - by using unsave generics, we are able to share one static instance return READ_ONLY_EMPTY_SET; } @Override - public @NonNull Long getOffsetHighestSeen() { + public long getOffsetHighestSeen() { log.debug(NO_OP); return -1L; } @Override - public long getOffsetHighestSucceeded() { // NOSONAR + public long getOffsetHighestSucceeded() { log.debug(NO_OP); return -1L; } - @Override - NavigableMap> getCommitQueue() { - //noinspection unchecked - by using unsave generics, we are able to share one static instance - return READ_ONLY_EMPTY_MAP; - } - - @Override - public void setIncompleteOffsets(final Set incompleteOffsets) { - log.debug(NO_OP); - } - - @Override - void setAllowedMoreRecords(final boolean allowedMoreRecords) { - log.debug(NO_OP); - } - - @Override - public void maybeRaiseHighestSeenOffset(final long highestSeen) { - log.debug(NO_OP); - } - - @Override - public void truncateOffsets(final long nextExpectedOffset) { - log.debug(NO_OP); - } - - @Override - public void onOffsetCommitSuccess(final OffsetAndMetadata committed) { - log.debug(NO_OP); - } - @Override public boolean isRecordPreviouslyCompleted(final ConsumerRecord rec) { log.debug("Ignoring previously completed request for partition no longer assigned. Partition: {}", KafkaUtils.toTopicPartition(rec)); diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkManager.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkManager.java index 71c9f71ba..fdd100016 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkManager.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkManager.java @@ -196,7 +196,8 @@ public List> getWorkIfAvailable(final int requestedMaxWorkTo * @return the number of extra records ingested due to request */ // todo rename - shunt messages from internal buffer into queues - int tryToEnsureQuantityOfWorkQueuedAvailable(final int requestedMaxWorkToRetrieve) { + // visible for testing + public int tryToEnsureQuantityOfWorkQueuedAvailable(final int requestedMaxWorkToRetrieve) { // todo this counts all partitions as a whole - this may cause some partitions to starve. need to round robin it? long available = sm.getNumberOfWorkQueuedInShardsAwaitingSelection(); long extraNeededFromInboxToSatisfy = requestedMaxWorkToRetrieve - available; @@ -238,6 +239,7 @@ public void onOffsetCommitSuccess(Map committ public void onFailureResult(WorkContainer wc) { // error occurred, put it back in the queue if it can be retried wc.endFlight(); + pm.onFailure(wc); sm.onFailure(wc); numberRecordsOutForProcessing--; } @@ -250,9 +252,8 @@ public Integer getAmountOfWorkQueuedWaitingIngestion() { return wmbm.getAmountOfWorkQueuedWaitingIngestion(); } - // todo rename - public Map findCompletedEligibleOffsetsAndRemove() { - return pm.findCompletedEligibleOffsetsAndRemove(); + public Map collectCommitDataForDirtyPartitions() { + return pm.collectDirtyCommitData(); } /** diff --git a/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/DbTest.java b/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/DbTest.java index 6e620da24..f0bcbb2cf 100644 --- a/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/DbTest.java +++ b/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/DbTest.java @@ -1,8 +1,8 @@ +package io.confluent.parallelconsumer.integrationTests; /*- - * Copyright (C) 2020-2021 Confluent, Inc. + * Copyright (C) 2020-2022 Confluent, Inc. */ -package io.confluent.parallelconsumer.integrationTests; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; @@ -19,14 +19,15 @@ import static org.assertj.core.api.Assertions.assertThat; /** - * Simulate real forward pressure, back pressure and error conditions by testing against a real database + * Simulate real forward pressure, back pressure and error conditions by testing against a real database, instead of + * just simulating "work" with a random sleep. */ @Slf4j public class DbTest extends BrokerIntegrationTest { protected static final PostgreSQLContainer dbc; - /** + /* * https://www.testcontainers.org/test_framework_integration/manual_lifecycle_control/#singleton-containers * https://github.com/testcontainers/testcontainers-java/pull/1781 */ diff --git a/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/LargeVolumeInMemoryTests.java b/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/LargeVolumeInMemoryTests.java index 167828bc2..f12d7f7b3 100644 --- a/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/LargeVolumeInMemoryTests.java +++ b/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/LargeVolumeInMemoryTests.java @@ -28,7 +28,6 @@ import java.util.*; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.CountDownLatch; -import java.util.stream.Collectors; import static io.confluent.csid.utils.GeneralTestUtils.time; import static io.confluent.csid.utils.Range.range; @@ -96,7 +95,7 @@ void load(CommitMode commitMode) { List> consumerCommitHistory = consumerSpy.getCommitHistoryInt(); assertThat(consumerCommitHistory).isNotEmpty(); - long mostRecentConsumerCommitOffset = consumerCommitHistory.get(consumerCommitHistory.size() - 1).values().stream().collect(Collectors.toList()).get(0).offset(); // non-tx + long mostRecentConsumerCommitOffset = new ArrayList<>(consumerCommitHistory.get(consumerCommitHistory.size() - 1).values()).get(0).offset(); // non-tx assertThat(mostRecentConsumerCommitOffset).isEqualTo(quantityOfMessagesToProduce); } diff --git a/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/VeryLargeMessageVolumeTest.java b/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/VeryLargeMessageVolumeTest.java index cdd658d2c..a4a7c292f 100644 --- a/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/VeryLargeMessageVolumeTest.java +++ b/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/VeryLargeMessageVolumeTest.java @@ -1,7 +1,7 @@ package io.confluent.parallelconsumer.integrationTests; /*- - * Copyright (C) 2020-2021 Confluent, Inc. + * Copyright (C) 2020-2022 Confluent, Inc. */ import io.confluent.csid.utils.ProgressBarUtils; @@ -62,14 +62,18 @@ public class VeryLargeMessageVolumeTest extends BrokerIntegrationTest * https://github.com/confluentinc/parallel-consumer/issues/35 + * + *

+ * See $37 Support BitSet and RunLength encoding lengths longer than Short.MAX_VALUE + *

+ * https://github.com/confluentinc/parallel-consumer/issues/37 *

- * https://github.com/confluentinc/parallel-consumer/issues/35 */ @Test - public void shouldNotThrowBitSetTooLongException() { + void shouldNotThrowBitSetTooLongException() { runTest(HIGH_MAX_POLL_RECORDS_CONFIG, CommitMode.PERIODIC_CONSUMER_ASYNCHRONOUS, ProcessingOrder.KEY); } @@ -81,7 +85,7 @@ private void runTest(int maxPoll, CommitMode commitMode, ProcessingOrder order) // pre-produce messages to input-topic List expectedKeys = new ArrayList<>(); // int expectedMessageCount = 2_000_000; - int expectedMessageCount = 100_0000; + long expectedMessageCount = 1_000_000; log.info("Producing {} messages before starting test", expectedMessageCount); List> sends = new ArrayList<>(); try (Producer kafkaProducer = kcu.createNewProducer(false)) { @@ -102,7 +106,7 @@ private void runTest(int maxPoll, CommitMode commitMode, ProcessingOrder order) for (Future send : sends) { send.get(); } - assertThat(sends).hasSize(expectedMessageCount); + assertThat(sends).hasSize((int) expectedMessageCount); // run parallel-consumer log.debug("Starting test"); @@ -112,12 +116,11 @@ private void runTest(int maxPoll, CommitMode commitMode, ProcessingOrder order) consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPoll); KafkaConsumer newConsumer = kcu.createNewConsumer(true, consumerProps); - var pc = new ParallelEoSStreamProcessor(ParallelConsumerOptions.builder() + var pc = new ParallelEoSStreamProcessor<>(ParallelConsumerOptions.builder() .ordering(order) .consumer(newConsumer) .producer(newProducer) .commitMode(commitMode) -// .numberOfThreads(1) .maxConcurrency(1000) .build()); pc.subscribe(of(inputName)); @@ -126,13 +129,13 @@ private void runTest(int maxPoll, CommitMode commitMode, ProcessingOrder order) TopicPartition tp = new TopicPartition(inputName, 0); Map beginOffsets = newConsumer.beginningOffsets(of(tp)); Map endOffsets = newConsumer.endOffsets(of(tp)); - assertThat(endOffsets.get(tp)).isEqualTo(expectedMessageCount); - assertThat(beginOffsets.get(tp)).isEqualTo(0L); + assertThat(endOffsets).containsEntry(tp, expectedMessageCount); + assertThat(beginOffsets).containsEntry(tp, 0L); ProgressBar bar = ProgressBarUtils.getNewMessagesBar(log, expectedMessageCount); pc.pollAndProduce(record -> { - // by not having any sleep here, this test really test the overhead of the system + // by not having any sleep here, this test really tests the overhead of the system // try { // Thread.sleep(5); // } catch (InterruptedException e) { diff --git a/parallel-consumer-core/src/test/java/io/confluent/csid/utils/KafkaTestUtils.java b/parallel-consumer-core/src/test/java/io/confluent/csid/utils/KafkaTestUtils.java index 4d3e0f423..98c8873ed 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/csid/utils/KafkaTestUtils.java +++ b/parallel-consumer-core/src/test/java/io/confluent/csid/utils/KafkaTestUtils.java @@ -41,6 +41,7 @@ public class KafkaTestUtils { int offset = 0; + // todo not used anymore - delete? public void assignConsumerToTopic(final MockConsumer mc) { TopicPartition tp1 = new TopicPartition(INPUT_TOPIC, 1); TopicPartition tp0 = new TopicPartition(INPUT_TOPIC, 0); @@ -56,13 +57,8 @@ public void assignConsumerToTopic(final MockConsumer mc) { * It's a race to see if the genesis offset gets committed or not. So lets remove it if it exists, and all tests can * assume it doesn't. */ - public static List trimAllGeneisOffset(final List collect) { - while (!collect.isEmpty() && collect.get(0) == 0) { - int genesisOffset = collect.get(0); - if (genesisOffset == 0) - collect.remove(0); - } - return collect; + public static List trimAllGenesisOffset(final List collect) { + return collect.stream().filter(x -> x != 0).collect(Collectors.toList()); } public ConsumerRecord makeRecord(String key, String value) { @@ -90,17 +86,17 @@ public void assertCommits(MockProducer mp, List integers) { */ public void assertCommits(MockProducer mp, List expectedOffsets, Optional description) { log.debug("Asserting commits of {}", expectedOffsets); - List set = getProducerCommits(mp); + List offsets = getProducerCommitsFlattened(mp); if (!expectedOffsets.contains(0)) { - KafkaTestUtils.trimAllGeneisOffset(set); + offsets = KafkaTestUtils.trimAllGenesisOffset(offsets); } - assertThat(set).describedAs(description.orElse("Which offsets are committed and in the expected order")) + assertThat(offsets).describedAs(description.orElse("Which offsets are committed and in the expected order")) .containsExactlyElementsOf(expectedOffsets); } - public List getProducerCommits(MockProducer mp) { + public List getProducerCommitsFlattened(MockProducer mp) { List>> history = mp.consumerGroupOffsetsHistory(); List set = history.stream().flatMap(histories -> { diff --git a/parallel-consumer-core/src/test/java/io/confluent/csid/utils/ProgressBarUtils.java b/parallel-consumer-core/src/test/java/io/confluent/csid/utils/ProgressBarUtils.java index 462b43091..d1ac18058 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/csid/utils/ProgressBarUtils.java +++ b/parallel-consumer-core/src/test/java/io/confluent/csid/utils/ProgressBarUtils.java @@ -1,7 +1,7 @@ package io.confluent.csid.utils; /*- - * Copyright (C) 2020-2021 Confluent, Inc. + * Copyright (C) 2020-2022 Confluent, Inc. */ import lombok.experimental.UtilityClass; @@ -13,24 +13,24 @@ @UtilityClass public class ProgressBarUtils { - public static ProgressBar getNewMessagesBar(Logger log, int initialMax) { + + public static ProgressBar getNewMessagesBar(Logger log, long initialMax) { return getNewMessagesBar(null, log, initialMax); } - public static ProgressBar getNewMessagesBar(String name, Logger log, int initialMax) { + public static ProgressBar getNewMessagesBar(String name, Logger log, long initialMax) { DelegatingProgressBarConsumer delegatingProgressBarConsumer = new DelegatingProgressBarConsumer(log::info); String usedName = "progress"; if (name != null) usedName = name; - ProgressBar build = new ProgressBarBuilder() + return new ProgressBarBuilder() .setConsumer(delegatingProgressBarConsumer) .setInitialMax(initialMax) .showSpeed() .setTaskName(usedName) .setUnit("msg", 1) .build(); - return build; } } diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/AbstractParallelEoSStreamProcessorTestBase.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/AbstractParallelEoSStreamProcessorTestBase.java index 82811d562..2988f5540 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/AbstractParallelEoSStreamProcessorTestBase.java +++ b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/AbstractParallelEoSStreamProcessorTestBase.java @@ -31,8 +31,8 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; +import java.util.stream.Stream; -import static io.confluent.csid.utils.KafkaTestUtils.trimAllGeneisOffset; import static io.confluent.csid.utils.LatchTestUtils.awaitLatch; import static io.confluent.csid.utils.StringUtils.msg; import static io.confluent.parallelconsumer.ParallelConsumerOptions.CommitMode.*; @@ -311,7 +311,7 @@ protected void awaitForCommitExact(int offset) { log.debug("Waiting for EXACTLY commit offset {}", offset); await().timeout(defaultTimeout) .failFast(msg("Commit was not exact - contained offsets that weren't '{}'", offset), () -> { - List offsets = extractAllPartitionsOffsetsSequentially(); + List offsets = extractAllPartitionsOffsetsSequentially(false); return offsets.size() > 1 && !offsets.contains(offset); }) .untilAsserted(() -> assertCommits(of(offset))); @@ -336,8 +336,8 @@ public void assertCommitsContains(List offsets) { private List getCommitHistoryFlattened() { return (isUsingTransactionalProducer()) - ? ktu.getProducerCommits(producerSpy) - : extractAllPartitionsOffsetsSequentially(); + ? ktu.getProducerCommitsFlattened(producerSpy) + : extractAllPartitionsOffsetsSequentially(false); } public void assertCommits(List offsets, String description) { @@ -349,27 +349,22 @@ public void assertCommits(List offsets, String description) { * exists, unless it's contained in the assertion. */ public void assertCommits(List offsets, Optional description) { + boolean trimGenesis = !offsets.contains(0); + if (isUsingTransactionalProducer()) { ktu.assertCommits(producerSpy, offsets, description); - assertThat(extractAllPartitionsOffsetsSequentially()).isEmpty(); + assertThat(extractAllPartitionsOffsetsSequentially(trimGenesis)).isEmpty(); } else { - List collect = extractAllPartitionsOffsetsSequentially(); - if (!offsets.contains(0)) { - collect = trimAllGeneisOffset(collect); - } + List collect = extractAllPartitionsOffsetsSequentially(trimGenesis); + // duplicates are ok // is there a nicer optional way? // {@link Optional#ifPresentOrElse} only @since 9 if (description.isPresent()) { assertThat(collect).as(description.get()).hasSameElementsAs(offsets); } else { - try { - assertThat(collect).hasSameElementsAs(offsets); - } catch (AssertionError e) { - throw e; - } + assertThat(collect).hasSameElementsAs(offsets); } - ktu.assertCommits(producerSpy, UniLists.of(), Optional.of("Empty")); } } @@ -377,14 +372,18 @@ public void assertCommits(List offsets, Optional description) { /** * Flattens the offsets of all partitions into a single sequential list */ - protected List extractAllPartitionsOffsetsSequentially() { + protected List extractAllPartitionsOffsetsSequentially(boolean trimGenesis) { // copy the list for safe concurrent access List> history = new ArrayList<>(consumerSpy.getCommitHistoryInt()); return history.stream() .flatMap(commits -> { Collection values = new ArrayList<>(commits.values()); // 4 debugging - return values.stream().map(meta -> (int) meta.offset()); // int cast a luxury in test context - no big offsets + Stream rawOffsets = values.stream().map(meta -> (int) meta.offset()); + if (trimGenesis) + return rawOffsets.filter(x -> x != 0); + else + return rawOffsets; // int cast a luxury in test context - no big offsets } ).collect(Collectors.toList()); } diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/BatchTestMethods.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/BatchTestMethods.java index 8ba88c8c5..39bf93c46 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/BatchTestMethods.java +++ b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/BatchTestMethods.java @@ -77,6 +77,7 @@ public void averageBatchSizeTest(int numRecsExpected) { // waitAtMost(defaultTimeout).alias("expected number of records") + .failFast(() -> getPC().isClosedOrFailed()) .untilAsserted(() -> { bar.stepTo(numRecordsProcessed.get()); assertThat(numRecordsProcessed.get()).isEqualTo(numRecsExpected); diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/ParallelEoSStreamProcessorTest.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/ParallelEoSStreamProcessorTest.java index 274520939..1214a9c78 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/ParallelEoSStreamProcessorTest.java +++ b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/ParallelEoSStreamProcessorTest.java @@ -384,7 +384,7 @@ void controlFlowException(CommitMode commitMode) { @ParameterizedTest() @EnumSource(CommitMode.class) @SneakyThrows - public void testVoidPollMethod(CommitMode commitMode) { + void testVoidPollMethod(CommitMode commitMode) { setupParallelConsumerInstance(commitMode); int expected = 1; @@ -396,7 +396,7 @@ public void testVoidPollMethod(CommitMode commitMode) { awaitLatch(msgCompleteBarrier); - awaitForOneLoopCycle(); + awaitForSomeLoopCycles(2); parallelConsumer.close(); diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/BitSetEncodingTest.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/BitSetEncodingTest.java index c22915398..eb6e3dc4a 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/BitSetEncodingTest.java +++ b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/BitSetEncodingTest.java @@ -1,7 +1,7 @@ package io.confluent.parallelconsumer.offsets; /*- - * Copyright (C) 2020-2021 Confluent, Inc. + * Copyright (C) 2020-2022 Confluent, Inc. */ import lombok.SneakyThrows; @@ -53,7 +53,7 @@ void basic() { OffsetMapCodecManager.HighestOffsetAndIncompletes result = OffsetMapCodecManager.decodeCompressedOffsets(0, wrapped); - assertThat(result.getHighestSeenOffset()).isEqualTo(10); + assertThat(result.getHighestSeenOffset()).contains(10L); assertThat(result.getIncompleteOffsets()).containsExactlyInAnyOrderElementsOf(incompletes); } diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetCodecTestUtils.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetCodecTestUtils.java new file mode 100644 index 000000000..83a5efb42 --- /dev/null +++ b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetCodecTestUtils.java @@ -0,0 +1,68 @@ +package io.confluent.parallelconsumer.offsets; + +/*- + * Copyright (C) 2020-2022 Confluent, Inc. + */ + +import io.confluent.parallelconsumer.state.PartitionState; +import lombok.experimental.UtilityClass; +import lombok.extern.slf4j.Slf4j; + +import java.util.HashSet; +import java.util.Set; + +import static io.confluent.csid.utils.Range.range; + +@Slf4j +@UtilityClass +public class OffsetCodecTestUtils { + + /** + * x is complete + *

+ * o is incomplete + */ + static String incompletesToBitmapString(long finalOffsetForPartition, long highestSeen, Set incompletes) { + var runLengthString = new StringBuilder(); + Long lowWaterMark = finalOffsetForPartition; + long end = highestSeen - lowWaterMark; + for (final var relativeOffset : range(end)) { + long offset = lowWaterMark + relativeOffset; + if (incompletes.contains(offset)) { + runLengthString.append("o"); + } else { + runLengthString.append("x"); + } + } + return runLengthString.toString(); + } + + static String incompletesToBitmapString(long finalOffsetForPartition, PartitionState state) { + return incompletesToBitmapString(finalOffsetForPartition, + state.getOffsetHighestSeen(), state.getIncompleteOffsetsBelowHighestSucceeded()); + } + + /** + * x is complete + *

+ * o is incomplete + */ + static Set bitmapStringToIncomplete(final long baseOffset, final String inputBitmapString) { + final Set incompleteOffsets = new HashSet<>(); + + final long longLength = inputBitmapString.length(); + range(longLength).forEach(i -> { + var bit = inputBitmapString.charAt(i); + if (bit == 'o') { + incompleteOffsets.add(baseOffset + i); + } else if (bit == 'x') { + log.trace("Dropping completed offset"); + } else { + throw new IllegalArgumentException("Invalid encoding - unexpected char: " + bit); + } + }); + + return incompleteOffsets; + } + +} diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingBackPressureTest.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingBackPressureTest.java index 19511f4cd..d8b8c5b03 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingBackPressureTest.java +++ b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingBackPressureTest.java @@ -5,9 +5,11 @@ */ import com.google.common.truth.Truth; +import com.google.common.truth.Truth8; import io.confluent.parallelconsumer.FakeRuntimeError; import io.confluent.parallelconsumer.ParallelEoSStreamProcessorTestBase; import io.confluent.parallelconsumer.offsets.OffsetMapCodecManager.HighestOffsetAndIncompletes; +import io.confluent.parallelconsumer.state.PartitionMonitor; import io.confluent.parallelconsumer.state.WorkContainer; import io.confluent.parallelconsumer.state.WorkManager; import lombok.extern.slf4j.Slf4j; @@ -137,7 +139,7 @@ void backPressureShouldPreventTooManyMessagesBeingQueuedForProcessing() throws O // check offset encoding incomplete payload doesn't contain expected completed messages String metadata = mostRecentCommit.metadata(); HighestOffsetAndIncompletes decodedOffsetPayload = OffsetMapCodecManager.deserialiseIncompleteOffsetMapFromBase64(0, metadata); - Long highestSeenOffset = decodedOffsetPayload.getHighestSeenOffset(); + Long highestSeenOffset = decodedOffsetPayload.getHighestSeenOffset().get(); Set incompletes = decodedOffsetPayload.getIncompleteOffsets(); assertThat(incompletes).isNotEmpty() .contains(offsetToBlock) @@ -189,7 +191,7 @@ void backPressureShouldPreventTooManyMessagesBeingQueuedForProcessing() throws O .deserialiseIncompleteOffsetMapFromBase64(0L, meta); Truth.assertWithMessage("The only incomplete record now is offset zero, which we are blocked on") .that(incompletes.getIncompleteOffsets()).containsExactlyElementsIn(blockedOffsets); - assertThat(incompletes.getHighestSeenOffset()).isEqualTo(numberOfRecords + extraRecordsToBlockWithThresholdBlocks - 1); + Truth8.assertThat(incompletes.getHighestSeenOffset()).hasValue(numberOfRecords + extraRecordsToBlockWithThresholdBlocks - 1); } ); } @@ -199,7 +201,7 @@ void backPressureShouldPreventTooManyMessagesBeingQueuedForProcessing() throws O int extraMessages = numberOfRecords + extraRecordsToBlockWithThresholdBlocks / 2; { // force system to allow more records (i.e. the actual system attempts to never allow the payload to grow this big) - wm.getPm().setUSED_PAYLOAD_THRESHOLD_MULTIPLIER(2); + PartitionMonitor.setUSED_PAYLOAD_THRESHOLD_MULTIPLIER(2); // msgLockThree.countDown(); // unlock to make state dirty to get a commit @@ -251,7 +253,7 @@ void backPressureShouldPreventTooManyMessagesBeingQueuedForProcessing() throws O // assert all committed, nothing blocked- next expected offset is now 1+ the offset of the final message we sent { await().untilAsserted(() -> { - List offsets = extractAllPartitionsOffsetsSequentially(); + List offsets = extractAllPartitionsOffsetsSequentially(false); assertThat(offsets).contains(userFuncFinishedCount.get()); }); await().untilAsserted(() -> assertThat(wm.getPm().isAllowedMoreRecords(topicPartition)).isTrue()); diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingTests.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingTests.java index 851558de5..0507a78ec 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingTests.java +++ b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingTests.java @@ -4,6 +4,7 @@ * Copyright (C) 2020-2022 Confluent, Inc. */ +import com.google.common.truth.Truth; import io.confluent.csid.utils.KafkaTestUtils; import io.confluent.parallelconsumer.ParallelConsumerOptions; import io.confluent.parallelconsumer.ParallelEoSStreamProcessorTestBase; @@ -19,16 +20,15 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.ValueSource; +import pl.tlinkowski.unij.api.UniLists; import pl.tlinkowski.unij.api.UniSets; import java.nio.ByteBuffer; import java.util.*; import java.util.stream.Collectors; -import static io.confluent.parallelconsumer.offsets.OffsetEncoding.ByteArray; -import static io.confluent.parallelconsumer.offsets.OffsetEncoding.ByteArrayCompressed; +import static io.confluent.parallelconsumer.offsets.OffsetEncoding.*; 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; @@ -42,7 +42,7 @@ public class OffsetEncodingTests extends ParallelEoSStreamProcessorTestBase { @Test void runLengthDeserialise() { var sb = ByteBuffer.allocate(3); - sb.put((byte) 0); // magic byte place holder, can ignore + sb.put((byte) 0); // magic byte placeholder, can ignore sb.putShort((short) 1); byte[] array = new byte[2]; sb.rewind(); @@ -127,6 +127,7 @@ void largeIncompleteOffsetValues(long nextExpectedOffset) { void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(OffsetEncoding encoding) { assumeThat("Codec skipped, not applicable", encoding, not(in(of(ByteArray, ByteArrayCompressed)))); // byte array not currently used + var encodingsThatFail = UniLists.of(BitSet, BitSetCompressed, BitSetV2, RunLength, RunLengthCompressed); // todo don't use static public accessors to change things - makes parallel testing harder and is smelly OffsetMapCodecManager.forcedCodec = Optional.of(encoding); @@ -158,22 +159,25 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset incompleteRecords.remove(incompleteRecords.stream().filter(x -> x.offset() == 25_000).findFirst().get()); incompleteRecords.remove(incompleteRecords.stream().filter(x -> x.offset() == highest).findFirst().get()); + List expected = incompleteRecords.stream().map(ConsumerRecord::offset) + .sorted() + .collect(Collectors.toList()); // ktu.send(consumerSpy, records); // - ParallelConsumerOptions options = parallelConsumer.getWm().getOptions(); + ParallelConsumerOptions options = parallelConsumer.getWm().getOptions(); HashMap>> recordsMap = new HashMap<>(); TopicPartition tp = new TopicPartition(INPUT_TOPIC, 0); recordsMap.put(tp, records); - ConsumerRecords crs = new ConsumerRecords<>(recordsMap); + ConsumerRecords testRecords = new ConsumerRecords<>(recordsMap); // write offsets { WorkManager wmm = new WorkManager<>(options, consumerSpy); wmm.onPartitionsAssigned(UniSets.of(new TopicPartition(INPUT_TOPIC, 0))); - wmm.registerWork(crs); + wmm.registerWork(testRecords); List> work = wmm.getWorkIfAvailable(); assertThat(work).hasSameSizeAs(records); @@ -187,42 +191,97 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset KafkaTestUtils.completeWork(wmm, work, highest); - var completedEligibleOffsetsAndRemove = wmm.findCompletedEligibleOffsetsAndRemove(); + // make the commit + var completedEligibleOffsets = wmm.collectCommitDataForDirtyPartitions(); + assertThat(completedEligibleOffsets.get(tp).offset()).isEqualTo(1L); + consumerSpy.commitSync(completedEligibleOffsets); - // check for graceful fall back to the smallest available encoder - OffsetMapCodecManager om = new OffsetMapCodecManager<>(consumerSpy); - OffsetMapCodecManager.forcedCodec = Optional.empty(); // turn off forced - var state = wmm.getPm().getPartitionState(tp); - String bestPayload = om.makeOffsetMetadataPayload(1, state); - assertThat(bestPayload).isNotEmpty(); + { + // check for graceful fall back to the smallest available encoder + OffsetMapCodecManager om = new OffsetMapCodecManager<>(consumerSpy); + OffsetMapCodecManager.forcedCodec = Optional.empty(); // turn off forced + var state = wmm.getPm().getPartitionState(tp); + String bestPayload = om.makeOffsetMetadataPayload(1, state); + assertThat(bestPayload).isNotEmpty(); + } + } + + // check + { + var committed = consumerSpy.committed(UniSets.of(tp)).get(tp); + assertThat(committed.offset()).isEqualTo(1L); - // - consumerSpy.commitSync(completedEligibleOffsetsAndRemove); + if (!encodingsThatFail.contains(encoding)) { + assertThat(committed.metadata()).isNotBlank(); + } } // simulate a rebalance or some sort of reset, by instantiating a new WM with the state from the last // read offsets { - WorkManager newWm = new WorkManager<>(options, consumerSpy); + var newWm = new WorkManager<>(options, consumerSpy); newWm.onPartitionsAssigned(UniSets.of(tp)); - newWm.registerWork(crs); - List> workRetrieved = newWm.getWorkIfAvailable(); + newWm.registerWork(testRecords); + + var pm = newWm.getPm(); + var partitionState = pm.getPartitionState(tp); + + if (!encodingsThatFail.contains(encoding)) { + long offsetHighestSequentialSucceeded = partitionState.getOffsetHighestSequentialSucceeded(); + assertThat(offsetHighestSequentialSucceeded).isEqualTo(0); + + long offsetHighestSucceeded = partitionState.getOffsetHighestSucceeded(); + assertThat(offsetHighestSucceeded).isEqualTo(highest); + + long offsetHighestSeen = partitionState.getOffsetHighestSeen(); + assertThat(offsetHighestSeen).isEqualTo(highest); + + var incompletes = partitionState.getIncompleteOffsetsBelowHighestSucceeded(); + Truth.assertThat(incompletes).containsExactlyElementsIn(expected); + } + + // check record is marked as incomplete + var anIncompleteRecord = records.get(3); + Truth.assertThat(pm.isRecordPreviouslyCompleted(anIncompleteRecord)).isFalse(); + + // force ingestion early, and check results + { + int ingested = newWm.tryToEnsureQuantityOfWorkQueuedAvailable(Integer.MAX_VALUE); + + if (!encodingsThatFail.contains(encoding)) { + long offsetHighestSequentialSucceeded = partitionState.getOffsetHighestSequentialSucceeded(); + assertThat(offsetHighestSequentialSucceeded).isEqualTo(0); + + long offsetHighestSucceeded = partitionState.getOffsetHighestSucceeded(); + assertThat(offsetHighestSucceeded).isEqualTo(highest); + + long offsetHighestSeen = partitionState.getOffsetHighestSeen(); + assertThat(offsetHighestSeen).isEqualTo(highest); + + var incompletes = partitionState.getIncompleteOffsetsBelowHighestSucceeded(); + Truth.assertThat(incompletes).containsExactlyElementsIn(expected); + + assertThat(ingested).isEqualTo(testRecords.count() - 4); // 4 were succeeded + Truth.assertThat(pm.isRecordPreviouslyCompleted(anIncompleteRecord)).isFalse(); + } + } + + + var workRetrieved = newWm.getWorkIfAvailable(); + var workRetrievedOffsets = workRetrieved.stream().map(WorkContainer::offset).collect(Collectors.toList()); + Truth.assertThat(workRetrieved).isNotEmpty(); + 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(workRetrieved).extracting(WorkContainer::getCr) - .containsExactlyElementsOf(incompleteRecords)) - .hasMessageContaining("but some elements were not") - .hasMessageContaining("offset = 25000"); + assertThat(workRetrievedOffsets).doesNotContain(2500L); + assertThat(workRetrievedOffsets).doesNotContainSequence(expected); } default -> { - List expected = incompleteRecords.stream().map(ConsumerRecord::offset).collect(Collectors.toList()); - Collections.sort(expected); - assertThat(workRetrieved.stream().map(WorkContainer::offset)) + assertThat(workRetrievedOffsets) .as("Contains only incomplete records") .containsExactlyElementsOf(expected); } diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/RunLengthEncoderTest.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/RunLengthEncoderTest.java index 8a941a9cc..f839cc495 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/RunLengthEncoderTest.java +++ b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/RunLengthEncoderTest.java @@ -1,7 +1,7 @@ package io.confluent.parallelconsumer.offsets; /*- - * Copyright (C) 2020-2021 Confluent, Inc. + * Copyright (C) 2020-2022 Confluent, Inc. */ import io.confluent.parallelconsumer.offsets.OffsetMapCodecManager.HighestOffsetAndIncompletes; @@ -96,7 +96,7 @@ void noGapsSerialisation() { HighestOffsetAndIncompletes result = OffsetMapCodecManager.decodeCompressedOffsets(0, wrapped); - assertThat(result.getHighestSeenOffset()).isEqualTo(10); + assertThat(result.getHighestSeenOffset()).contains(10L); assertThat(result.getIncompleteOffsets()).containsExactlyElementsOf(incompletes); } diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/WorkManagerOffsetMapCodecManagerTest.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/WorkManagerOffsetMapCodecManagerTest.java index 538abad9f..dc83ad39d 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/WorkManagerOffsetMapCodecManagerTest.java +++ b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/WorkManagerOffsetMapCodecManagerTest.java @@ -1,16 +1,20 @@ package io.confluent.parallelconsumer.offsets; /*- - * Copyright (C) 2020-2021 Confluent, Inc. + * Copyright (C) 2020-2022 Confluent, Inc. */ +import com.google.common.truth.Truth; +import io.confluent.csid.utils.TimeUtils; import io.confluent.parallelconsumer.ParallelConsumerOptions; import io.confluent.parallelconsumer.state.PartitionState; import io.confluent.parallelconsumer.state.WorkContainer; import io.confluent.parallelconsumer.state.WorkManager; +import lombok.Getter; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.RandomUtils; +import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.MockConsumer; import org.apache.kafka.clients.consumer.OffsetResetStrategy; import org.apache.kafka.common.TopicPartition; @@ -18,7 +22,12 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.Mock; import org.mockito.Mockito; +import org.mockito.junit.jupiter.MockitoExtension; import org.xerial.snappy.SnappyOutputStream; import pl.tlinkowski.unij.api.UniLists; import pl.tlinkowski.unij.api.UniMaps; @@ -29,22 +38,23 @@ import java.nio.ByteBuffer; import java.util.*; +import static com.google.common.truth.Truth.assertWithMessage; import static io.confluent.csid.utils.Range.range; +import static io.confluent.parallelconsumer.offsets.OffsetCodecTestUtils.bitmapStringToIncomplete; +import static io.confluent.parallelconsumer.offsets.OffsetCodecTestUtils.incompletesToBitmapString; import static io.confluent.parallelconsumer.offsets.OffsetEncoding.*; import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.Optional.of; import static org.assertj.core.api.Assertions.assertThat; -/** - * TODO: use compressed avro instead for a more reliable long term schema system? or string encoding with a version - * prefix fine? - */ // todo refactor - remove tests which use hard coded state vs dynamic state - #compressionCycle, #selialiseCycle, #runLengthEncoding, #loadCompressedRunLengthRncoding @Slf4j +@ExtendWith(MockitoExtension.class) class WorkManagerOffsetMapCodecManagerTest { WorkManager wm; - OffsetMapCodecManager om; + OffsetMapCodecManager offsetCodecManager; TopicPartition tp = new TopicPartition("myTopic", 0); @@ -54,21 +64,32 @@ class WorkManagerOffsetMapCodecManagerTest { TreeSet incompleteOffsets = new TreeSet<>(UniSets.of(0L, 2L, 3L)); /** - * Committable offset of 2, meaning 1 is complete and 2 and 3 are incomplete. 4 is also complete. + * Committable offset of 0, meaning 1 and 4 are complete and 2 and 3 are incomplete + *

+ * 0X00X */ long finalOffsetForPartition = 0L; /** * Sample data runs up to a highest seen offset of 4. Where offset 3 and 3 are incomplete. */ - long partitionHighWaterMark = 4; + long highestSucceeded = 4; + + PartitionState state = new PartitionState<>(tp, new OffsetMapCodecManager.HighestOffsetAndIncompletes(of(highestSucceeded), incompleteOffsets)); - PartitionState state = new PartitionState<>(tp, new OffsetMapCodecManager.HighestOffsetAndIncompletes(partitionHighWaterMark, incompleteOffsets)); + @Mock + ConsumerRecord mockCr; - { - WorkContainer mock = Mockito.mock(WorkContainer.class); - Mockito.doReturn(partitionHighWaterMark).when(mock).offset(); - state.onSuccess(mock); // in this case the highest seen is also the highest succeeded + @BeforeEach + void setupMock() { + injectSucceededWorkAtOffset(highestSucceeded); + } + + private void injectSucceededWorkAtOffset(long offset) { + WorkContainer workContainer = new WorkContainer<>(0, mockCr, null, TimeUtils.getClock()); + Mockito.doReturn(offset).when(mockCr).offset(); + state.addWorkContainer(workContainer); + state.onSuccess(workContainer); // in this case the highest seen is also the highest succeeded } /** @@ -91,6 +112,7 @@ class WorkManagerOffsetMapCodecManagerTest { "xxxxxxoooooxoxoxoooooxxxxoooooxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxoxoxooxoxoxoxoxoxoxoxoxoxoxoxo" ); + @Getter static List inputsToCompress = new ArrayList<>(); @BeforeEach @@ -98,12 +120,7 @@ void setup() { MockConsumer consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST); wm = new WorkManager<>(ParallelConsumerOptions.builder().build(), consumer); wm.onPartitionsAssigned(UniLists.of(tp)); - om = new OffsetMapCodecManager<>(consumer); - } - - // todo refactor tests out that depend on this - private void raiseToHardCodedHighestSeenOffset() { - wm.getPm().maybeRaiseHighestSeenOffset(tp, partitionHighWaterMark); + offsetCodecManager = new OffsetMapCodecManager<>(consumer); } @BeforeAll @@ -118,6 +135,7 @@ static void data() { inputsToCompress.add(input100 + input100 + input100 + input100 + input100 + input100 + input100 + input100 + input100 + input100 + input100); inputsToCompress.add(inputString); inputsToCompress.add(generateRandomData(1000).toString()); + // remove? slow, not needed? inputsToCompress.add(generateRandomData(10000).toString()); inputsToCompress.add(generateRandomData(30000).toString()); } @@ -133,7 +151,7 @@ private static StringBuffer generateRandomData(int entries) { @SneakyThrows @Test void serialiseCycle() { - String serialised = om.serialiseIncompleteOffsetMapToBase64(finalOffsetForPartition, state); + String serialised = offsetCodecManager.serialiseIncompleteOffsetMapToBase64(finalOffsetForPartition, state); log.info("Size: {}", serialised.length()); // @@ -240,30 +258,30 @@ void base64Encoding() { @SneakyThrows @Test void loadCompressedRunLengthEncoding() { - byte[] bytes = om.encodeOffsetsCompressed(finalOffsetForPartition, state); + byte[] bytes = offsetCodecManager.encodeOffsetsCompressed(finalOffsetForPartition, state); OffsetMapCodecManager.HighestOffsetAndIncompletes longs = OffsetMapCodecManager.decodeCompressedOffsets(finalOffsetForPartition, bytes); assertThat(longs.getIncompleteOffsets().toArray()).containsExactly(incompleteOffsets.toArray()); } @Test void decodeOffsetMap() { - Set set = OffsetMapCodecManager.bitmapStringToIncomplete(2L, "ooxx"); + Set set = bitmapStringToIncomplete(2L, "ooxx"); assertThat(set).containsExactly(2L, 3L); - assertThat(OffsetMapCodecManager.bitmapStringToIncomplete(2L, "ooxxoxox")).containsExactly(2L, 3L, 6L, 8L); - assertThat(OffsetMapCodecManager.bitmapStringToIncomplete(2L, "o")).containsExactly(2L); - assertThat(OffsetMapCodecManager.bitmapStringToIncomplete(2L, "x")).containsExactly(); - assertThat(OffsetMapCodecManager.bitmapStringToIncomplete(2L, "")).containsExactly(); - assertThat(OffsetMapCodecManager.bitmapStringToIncomplete(2L, "ooo")).containsExactly(2L, 3L, 4L); - assertThat(OffsetMapCodecManager.bitmapStringToIncomplete(2L, "xxx")).containsExactly(); + assertThat(bitmapStringToIncomplete(2L, "ooxxoxox")).containsExactly(2L, 3L, 6L, 8L); + assertThat(bitmapStringToIncomplete(2L, "o")).containsExactly(2L); + assertThat(bitmapStringToIncomplete(2L, "x")).containsExactly(); + assertThat(bitmapStringToIncomplete(2L, "")).containsExactly(); + assertThat(bitmapStringToIncomplete(2L, "ooo")).containsExactly(2L, 3L, 4L); + assertThat(bitmapStringToIncomplete(2L, "xxx")).containsExactly(); } @Test void binaryArrayConstruction() { - state.maybeRaiseHighestSeenOffset(6L); + injectSucceededWorkAtOffset(6); - String s = om.incompletesToBitmapString(1L, state); //2,3 - assertThat(s).isEqualTo("xooxx"); + String encoding = incompletesToBitmapString(finalOffsetForPartition, state); + assertThat(encoding).isEqualTo("oxooxx"); } @SneakyThrows @@ -298,8 +316,8 @@ void compressDecompressSanityZstd() { @SneakyThrows @Test void largeOffsetMap() { - state.maybeRaiseHighestSeenOffset(200L); // force system to have seen a high offset - byte[] bytes = om.encodeOffsetsCompressed(0L, state); + injectSucceededWorkAtOffset(200); // force system to have seen a high offset + byte[] bytes = offsetCodecManager.encodeOffsetsCompressed(0L, state); int smallestCompressionObserved = 10; assertThat(bytes).as("very small") .hasSizeLessThan(smallestCompressionObserved); // arbitrary size expectation based on past observations - expect around 7 @@ -311,9 +329,9 @@ void stringVsByteVsBitSetEncoding() { for (var inputString : inputsToCompress) { int inputLength = inputString.length(); - Set longs = OffsetMapCodecManager.bitmapStringToIncomplete(finalOffsetForPartition, inputString); + Set longs = bitmapStringToIncomplete(finalOffsetForPartition, inputString); - OffsetSimultaneousEncoder simultaneousEncoder = new OffsetSimultaneousEncoder(finalOffsetForPartition, partitionHighWaterMark, longs).invoke(); + OffsetSimultaneousEncoder simultaneousEncoder = new OffsetSimultaneousEncoder(finalOffsetForPartition, highestSucceeded, longs).invoke(); byte[] byteByte = simultaneousEncoder.getEncodingMap().get(ByteArray); byte[] bitsBytes = simultaneousEncoder.getEncodingMap().get(BitSet); @@ -335,7 +353,7 @@ void deserialiseBitSet() { long highestSucceeded = input.length() - 1; int nextExpectedOffset = 0; - Set incompletes = OffsetMapCodecManager.bitmapStringToIncomplete(nextExpectedOffset, input); + Set incompletes = bitmapStringToIncomplete(nextExpectedOffset, input); OffsetSimultaneousEncoder encoder = new OffsetSimultaneousEncoder(nextExpectedOffset, highestSucceeded, incompletes); encoder.invoke(); byte[] pack = encoder.packSmallest(); @@ -349,7 +367,7 @@ void deserialiseBitSet() { @SneakyThrows @Test void compressionCycle() { - byte[] serialised = om.encodeOffsetsCompressed(finalOffsetForPartition, state); + byte[] serialised = offsetCodecManager.encodeOffsetsCompressed(finalOffsetForPartition, state); OffsetMapCodecManager.HighestOffsetAndIncompletes deserialised = OffsetMapCodecManager.decodeCompressedOffsets(finalOffsetForPartition, serialised); @@ -358,58 +376,54 @@ void compressionCycle() { @Test void runLengthEncoding() { - raiseToHardCodedHighestSeenOffset(); - - String stringMap = om.incompletesToBitmapString(finalOffsetForPartition, state); + String stringMap = incompletesToBitmapString(finalOffsetForPartition, state); List integers = OffsetRunLength.runLengthEncode(stringMap); assertThat(integers).as("encoding of map: " + stringMap).containsExactlyElementsOf(UniLists.of(1, 1, 2)); assertThat(OffsetRunLength.runLengthDecodeToString(integers)).isEqualTo(stringMap); } + static List differentInputsAndCompressions() { + return inputsToCompress; + } + /** * Compare compression performance on different types of inputs, and tests that each encoding type is decompressed * again correctly */ - @Test - void differentInputsAndCompressions() { - for (final String input : inputsToCompress) { - // override high water mark setup, as the test sets it manually - setup(); - wm.getPm().maybeRaiseHighestSeenOffset(tp, 0L); // hard reset to zero - long highWater = input.length(); - wm.getPm().maybeRaiseHighestSeenOffset(tp, highWater); + @ParameterizedTest + @MethodSource + void differentInputsAndCompressions(String input) { + long highestSeen = input.length() - 1; // pretend we've gone one higher than the input incompletes + + // + log.debug("Testing round - size: {} input: '{}'", input.length(), input); + Set inputIncompletes = bitmapStringToIncomplete(finalOffsetForPartition, input); + String sanityEncoding = incompletesToBitmapString(finalOffsetForPartition, highestSeen + 1, inputIncompletes); + Truth.assertThat(sanityEncoding).isEqualTo(input); + // + OffsetSimultaneousEncoder encoder = new OffsetSimultaneousEncoder(finalOffsetForPartition, highestSeen, inputIncompletes); + encoder.invoke(); + + // test all encodings created + for (final EncodedOffsetPair encoding : encoder.sortedEncodings) { // - log.debug("Testing round - size: {} input: '{}'", input.length(), input); - Set longs = OffsetMapCodecManager.bitmapStringToIncomplete(finalOffsetForPartition, input); - OffsetSimultaneousEncoder encoder = new OffsetSimultaneousEncoder(finalOffsetForPartition, highWater, longs); - encoder.invoke(); - - // test all encodings created - for (final EncodedOffsetPair pair : encoder.sortedEncodings) { - byte[] result = encoder.packEncoding(pair); - - // - OffsetMapCodecManager.HighestOffsetAndIncompletes recoveredIncompleteOffsetTuple = - OffsetMapCodecManager.decodeCompressedOffsets(finalOffsetForPartition, result); - Set recoveredIncompletes = recoveredIncompleteOffsetTuple.getIncompleteOffsets(); - - // - assertThat(recoveredIncompletes).containsExactlyInAnyOrderElementsOf(longs); - - // - var state = new PartitionState(tp, new OffsetMapCodecManager.HighestOffsetAndIncompletes(highWater, recoveredIncompletes)); - String recoveredOffsetBitmapAsString = om.incompletesToBitmapString(finalOffsetForPartition, state); - assertThat(recoveredOffsetBitmapAsString).isEqualTo(input); - } - } - } + byte[] packedEncoding = encoder.packEncoding(encoding); - @Disabled - @Test - void testAllInputsEachEncoding() { - assertThat(true).isFalse(); + // + var recoveredIncompleteAndOffset = + OffsetMapCodecManager.decodeCompressedOffsets(finalOffsetForPartition, packedEncoding); + Set recoveredIncompletes = recoveredIncompleteAndOffset.getIncompleteOffsets(); + + // + assertThat(recoveredIncompletes).containsExactlyInAnyOrderElementsOf(inputIncompletes); + + // + String simple = incompletesToBitmapString(finalOffsetForPartition, highestSeen + 1, recoveredIncompletes); + assertWithMessage(encoding.encoding.name()) + .that(simple).isEqualTo(input); + } } } diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/state/WorkManagerTest.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/state/WorkManagerTest.java index 0f38cdb9d..0b08ab775 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/state/WorkManagerTest.java +++ b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/state/WorkManagerTest.java @@ -19,6 +19,7 @@ import org.apache.kafka.common.TopicPartition; import org.assertj.core.api.AbstractListAssert; import org.assertj.core.api.ObjectAssert; +import org.jetbrains.annotations.NotNull; import org.junit.jupiter.api.Assumptions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Disabled; @@ -74,7 +75,12 @@ private void setupWorkManager(ParallelConsumerOptions build) { } private void assignPartition(final int partition) { - wm.onPartitionsAssigned(UniLists.of(new TopicPartition(INPUT_TOPIC, partition))); + wm.onPartitionsAssigned(UniLists.of(getTopicPartition(partition))); + } + + @NotNull + private TopicPartition getTopicPartition(int partition) { + return new TopicPartition(INPUT_TOPIC, partition); } /** @@ -87,7 +93,7 @@ private void registerSomeWork() { var rec1 = makeRec("1", key, partition); var rec2 = makeRec("2", key, partition); Map>> m = new HashMap<>(); - m.put(new TopicPartition(INPUT_TOPIC, partition), of(rec0, rec1, rec2)); + m.put(getTopicPartition(partition), of(rec0, rec1, rec2)); var recs = new ConsumerRecords<>(m); wm.registerWork(recs); } @@ -177,6 +183,10 @@ private void succeed(WorkContainer succeed) { wm.onSuccessResult(succeed); } + private void succeed(Iterable> succeed) { + succeed.forEach(this::succeed); + } + /** * Checks the offsets of the work, matches the offsets in the provided list * @@ -326,7 +336,7 @@ void insertWrongOrderPreservesOffsetOrdering() { var rec2 = new ConsumerRecord<>(INPUT_TOPIC, partition, 6, key, "value"); var rec3 = new ConsumerRecord<>(INPUT_TOPIC, partition, 8, key, "value"); Map>> m = new HashMap<>(); - m.put(new TopicPartition(INPUT_TOPIC, partition), of(rec2, rec3, rec)); + m.put(getTopicPartition(partition), of(rec2, rec3, rec)); var recs = new ConsumerRecords<>(m); // @@ -435,7 +445,7 @@ void orderedByPartitionsParallel() { var rec2 = new ConsumerRecord<>(INPUT_TOPIC, partition, 6, "66", "value"); var rec3 = new ConsumerRecord<>(INPUT_TOPIC, partition, 8, "66", "value"); Map>> m = new HashMap<>(); - m.put(new TopicPartition(INPUT_TOPIC, partition), of(rec2, rec3, rec)); + m.put(getTopicPartition(partition), of(rec2, rec3, rec)); var recs = new ConsumerRecords<>(m); // @@ -481,7 +491,7 @@ void orderedByKeyParallel() { var rec5 = new ConsumerRecord<>(INPUT_TOPIC, partition, 15, "key-a", "value"); var rec6 = new ConsumerRecord<>(INPUT_TOPIC, partition, 20, "key-c", "value"); Map>> m = new HashMap<>(); - m.put(new TopicPartition(INPUT_TOPIC, partition), of(rec2, rec3, rec0, rec4, rec5, rec6)); + m.put(getTopicPartition(partition), of(rec2, rec3, rec0, rec4, rec5, rec6)); var recs = new ConsumerRecords<>(m); // @@ -533,7 +543,7 @@ void highVolumeKeyOrder(int quantity) { flattened.sort(comparingLong(ConsumerRecord::offset)); Map>> m = new HashMap<>(); - m.put(new TopicPartition(INPUT_TOPIC, 0), flattened); + m.put(getTopicPartition(0), flattened); var recs = new ConsumerRecords<>(m); // @@ -569,8 +579,8 @@ void treeMapOrderingCorrect() { * Checks work management is correct in this respect. */ @Test - public void workQueuesEmptyWhenAllWorkComplete() { - ParallelConsumerOptions build = ParallelConsumerOptions.builder() + void workQueuesEmptyWhenAllWorkComplete() { + var build = ParallelConsumerOptions.builder() .ordering(UNORDERED) .build(); setupWorkManager(build); @@ -581,18 +591,20 @@ public void workQueuesEmptyWhenAllWorkComplete() { assertThat(work).hasSize(3); // - for (var w : work) { - succeed(w); - } + succeed(work); // assertThat(wm.getSm().getNumberOfWorkQueuedInShardsAwaitingSelection()).isZero(); - assertThat(wm.getNumberOfEntriesInPartitionQueues()).isEqualTo(3); + assertThat(wm.getNumberOfEntriesInPartitionQueues()).as("Partition commit queues are now empty").isZero(); // drain commit queue - var completedFutureOffsets = wm.findCompletedEligibleOffsetsAndRemove(); + var completedFutureOffsets = wm.collectCommitDataForDirtyPartitions(); assertThat(completedFutureOffsets).hasSize(1); // coalesces (see log) - assertThat(wm.getNumberOfEntriesInPartitionQueues()).isEqualTo(0); + var sync = completedFutureOffsets.values().stream().findFirst().get(); + Truth.assertThat(sync.offset()).isEqualTo(3); + Truth.assertThat(sync.metadata()).isEmpty(); + PartitionState state = wm.getPm().getPartitionState(getTopicPartition(0)); + Truth.assertThat(state.getAllIncompleteOffsets()).isEmpty(); } /** @@ -614,9 +626,9 @@ void resumesFromNextShard(ParallelConsumerOptions.ProcessingOrder order) { assignPartition(2); Map>> m = new HashMap<>(); var rec = new ConsumerRecord<>(INPUT_TOPIC, 1, 11, "11", "value"); - m.put(new TopicPartition(INPUT_TOPIC, 1), of(rec)); + m.put(getTopicPartition(1), of(rec)); var rec2 = new ConsumerRecord<>(INPUT_TOPIC, 2, 21, "21", "value"); - m.put(new TopicPartition(INPUT_TOPIC, 2), of(rec2)); + m.put(getTopicPartition(2), of(rec2)); var recs = new ConsumerRecords<>(m); wm.registerWork(recs); diff --git a/parallel-consumer-core/src/test/resources/logback-test.xml b/parallel-consumer-core/src/test/resources/logback-test.xml index 62ccb809d..906410547 100644 --- a/parallel-consumer-core/src/test/resources/logback-test.xml +++ b/parallel-consumer-core/src/test/resources/logback-test.xml @@ -37,8 +37,8 @@ - - + + @@ -48,7 +48,7 @@ - + diff --git a/pom.xml b/pom.xml index 67e317c0a..47c5e9667 100644 --- a/pom.xml +++ b/pom.xml @@ -111,6 +111,7 @@ 1.16.3 1.1.3 0.7.4 + 4.4.0 @@ -235,6 +236,7 @@ + com.github.bsideup.jabel jabel-javac-plugin @@ -263,7 +265,7 @@ compile - + ch.qos.logback logback-classic @@ -302,9 +304,14 @@ org.mockito mockito-core - 4.4.0 + ${mockito.version} test + + org.mockito + mockito-junit-jupiter + ${mockito.version} + com.google.auto.service auto-service-annotations @@ -326,6 +333,16 @@ awaitility test + + com.google.flogger + flogger + test + + + com.google.flogger + flogger-slf4j-backend + test +