diff --git a/.idea/runConfigurations/generate_test_sources__e.xml b/.idea/runConfigurations/generate_test_sources__e.xml
new file mode 100644
index 000000000..2456e1d82
--- /dev/null
+++ b/.idea/runConfigurations/generate_test_sources__e.xml
@@ -0,0 +1,30 @@
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/parallel-consumer-core/src/main/java/io/confluent/csid/utils/MathUtils.java b/parallel-consumer-core/src/main/java/io/confluent/csid/utils/MathUtils.java
deleted file mode 100644
index 3bcfd0012..000000000
--- a/parallel-consumer-core/src/main/java/io/confluent/csid/utils/MathUtils.java
+++ /dev/null
@@ -1,22 +0,0 @@
-package io.confluent.csid.utils;
-
-/*-
- * Copyright (C) 2020-2022 Confluent, Inc.
- */
-
-public class MathUtils {
-
- /**
- * @deprecated old, was used with AssertJ - not needed with Truth (has fuzzy matching built in)
- */
- @Deprecated
- public static boolean isLessWithin(final int needle, final int target, final int percent) {
- int i = target * (1 - percent / 100);
-
- int diff = target - needle;
- int norm = Math.abs(diff);
- int off = norm / target * 100;
- return off < percent;
- }
-
-}
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RetriableException.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RetriableException.java
new file mode 100644
index 000000000..5e6ed1732
--- /dev/null
+++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RetriableException.java
@@ -0,0 +1,29 @@
+package io.confluent.parallelconsumer;
+
+/*-
+ * Copyright (C) 2020-2022 Confluent, Inc.
+ */
+
+/**
+ * A user's processing function can throw this exception, which signals to PC that processing of the message has failed,
+ * and that it should be retired at a later time.
+ *
+ * The advantage of throwing this exception explicitly, is that PC will not log an ERROR. If any other type of exception
+ * is thrown by the user's function, that will be logged as an error (but will still be retried later).
+ *
+ * So in short, if this exception is thrown, nothing will be logged (except at DEBUG level), any other exception will be
+ * logged as an error.
+ */
+public class RetriableException extends RuntimeException {
+ public RetriableException(String message) {
+ super(message);
+ }
+
+ public RetriableException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public RetriableException(Throwable cause) {
+ super(cause);
+ }
+}
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java
index ff9747d6d..71a6d472a 100644
--- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java
+++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java
@@ -8,12 +8,10 @@
import io.confluent.parallelconsumer.ParallelConsumer;
import io.confluent.parallelconsumer.ParallelConsumerOptions;
import io.confluent.parallelconsumer.PollContextInternal;
+import io.confluent.parallelconsumer.RetriableException;
import io.confluent.parallelconsumer.state.WorkContainer;
import io.confluent.parallelconsumer.state.WorkManager;
-import lombok.AccessLevel;
-import lombok.Getter;
-import lombok.Setter;
-import lombok.SneakyThrows;
+import lombok.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -46,6 +44,7 @@
import static java.time.Duration.ofMillis;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
+import static lombok.AccessLevel.PRIVATE;
import static lombok.AccessLevel.PROTECTED;
/**
@@ -55,6 +54,7 @@
public abstract class AbstractParallelEoSStreamProcessor implements ParallelConsumer, ConsumerRebalanceListener, Closeable {
public static final String MDC_INSTANCE_ID = "pcId";
+ public static final String MDC_OFFSET_MARKER = "offset";
@Getter(PROTECTED)
protected final ParallelConsumerOptions options;
@@ -99,7 +99,35 @@ public abstract class AbstractParallelEoSStreamProcessor implements Parall
* Collection of work waiting to be
*/
@Getter(PROTECTED)
- private final BlockingQueue> workMailBox = new LinkedBlockingQueue<>(); // Thread safe, highly performant, non blocking
+ private final BlockingQueue> workMailBox = new LinkedBlockingQueue<>(); // Thread safe, highly performant, non blocking
+
+ /**
+ * An inbound message to the controller.
+ *
+ * Currently, an Either type class, representing either newly polled records to ingest, or a work result.
+ */
+ @Value
+ @RequiredArgsConstructor(access = PRIVATE)
+ private static class ControllerEventMessage {
+ WorkContainer workContainer;
+ EpochAndRecordsMap consumerRecords;
+
+ private boolean isWorkResult() {
+ return workContainer != null;
+ }
+
+ private boolean isNewConsumerRecords() {
+ return !isWorkResult();
+ }
+
+ private static ControllerEventMessage of(EpochAndRecordsMap polledRecords) {
+ return new ControllerEventMessage<>(null, polledRecords);
+ }
+
+ public static ControllerEventMessage of(WorkContainer work) {
+ return new ControllerEventMessage(work, null);
+ }
+ }
private final BrokerPollSystem brokerPollSubsystem;
@@ -496,7 +524,13 @@ private boolean isResponsibleForCommits() {
/**
* Block the calling thread until no more messages are being processed.
+ *
+ * Used for testing.
+ *
+ * @deprecated no longer used, will be removed in next version
*/
+ // TODO delete
+ @Deprecated
@SneakyThrows
public void waitForProcessedNotCommitted(Duration timeout) {
log.debug("Waiting processed but not committed...");
@@ -621,9 +655,8 @@ private void controlLoop(Function, List> userFu
if (state == running) {
if (!wm.isSufficientlyLoaded() & brokerPollSubsystem.isPaused()) {
- // can occur
log.debug("Found Poller paused with not enough front loaded messages, ensuring poller is awake (mail: {} vs target: {})",
- wm.getAmountOfWorkQueuedWaitingIngestion(),
+ wm.getNumberOfWorkQueuedInShardsAwaitingSelection(),
options.getTargetAmountOfRecordsInFlight());
brokerPollSubsystem.wakeupIfPaused();
}
@@ -665,7 +698,7 @@ private void controlLoop(Function, List> userFu
// end of loop
log.trace("End of control loop, waiting processing {}, remaining in partition queues: {}, out for processing: {}. In state: {}",
- wm.getTotalWorkAwaitingIngestion(), wm.getNumberOfEntriesInPartitionQueues(), wm.getNumberRecordsOutForProcessing(), state);
+ wm.getNumberOfWorkQueuedInShardsAwaitingSelection(), wm.getNumberOfEntriesInPartitionQueues(), wm.getNumberRecordsOutForProcessing(), state);
}
private int handleWork(final Function, List> userFunction, final Consumer callback) {
@@ -812,20 +845,16 @@ protected int getQueueTargetLoaded() {
* Checks the system has enough pressure in the pipeline of work, if not attempts to step up the load factor.
*/
protected void checkPipelinePressure() {
- boolean moreWorkInQueuesAvailableThatHaveNotBeenPulled = wm.getAmountOfWorkQueuedWaitingIngestion() > options.getTargetAmountOfRecordsInFlight();
if (log.isTraceEnabled())
log.trace("Queue pressure check: (current size: {}, loaded target: {}, factor: {}) " +
- "if (isPoolQueueLow() {} && moreWorkInQueuesAvailableThatHaveNotBeenPulled {} && lastWorkRequestWasFulfilled {}))",
+ "if (isPoolQueueLow() {} && lastWorkRequestWasFulfilled {}))",
getNumberOfUserFunctionsQueued(),
getQueueTargetLoaded(),
dynamicExtraLoadFactor.getCurrentFactor(),
isPoolQueueLow(),
- moreWorkInQueuesAvailableThatHaveNotBeenPulled,
lastWorkRequestWasFulfilled);
- if (isPoolQueueLow()
- && moreWorkInQueuesAvailableThatHaveNotBeenPulled
- && lastWorkRequestWasFulfilled) {
+ if (isPoolQueueLow() && lastWorkRequestWasFulfilled) {
boolean steppedUp = dynamicExtraLoadFactor.maybeStepUp();
if (steppedUp) {
log.debug("isPoolQueueLow(): Executor pool queue is not loaded with enough work (queue: {} vs target: {}), stepped up loading factor to {}",
@@ -847,10 +876,9 @@ private boolean isPoolQueueLow() {
int queueSize = getNumberOfUserFunctionsQueued();
int queueTarget = getPoolLoadTarget();
boolean workAmountBelowTarget = queueSize <= queueTarget;
- boolean hasWorkInMailboxes = wm.hasWorkAwaitingIngestionToShards();
- log.debug("isPoolQueueLow()? workAmountBelowTarget {} {} vs {} && wm.hasWorkInMailboxes() {};",
- workAmountBelowTarget, queueSize, queueTarget, hasWorkInMailboxes);
- return workAmountBelowTarget && hasWorkInMailboxes;
+ log.debug("isPoolQueueLow()? workAmountBelowTarget {} {} vs {};",
+ workAmountBelowTarget, queueSize, queueTarget);
+ return workAmountBelowTarget;
}
private void drain() {
@@ -880,7 +908,7 @@ private void transitionToClosing() {
*/
private void processWorkCompleteMailBox() {
log.trace("Processing mailbox (might block waiting for results)...");
- Set> results = new HashSet<>();
+ Queue> results = new ArrayDeque<>();
final Duration timeToBlockFor = getTimeToBlockFor();
@@ -893,7 +921,7 @@ private void processWorkCompleteMailBox() {
// wait for work, with a timeToBlockFor for sanity
log.trace("Blocking poll {}", timeToBlockFor);
try {
- WorkContainer firstBlockingPoll = workMailBox.poll(timeToBlockFor.toMillis(), MILLISECONDS);
+ var firstBlockingPoll = workMailBox.poll(timeToBlockFor.toMillis(), MILLISECONDS);
if (firstBlockingPoll == null) {
log.debug("Mailbox results returned null, indicating timeToBlockFor (which was set as {})", timeToBlockFor);
} else {
@@ -915,10 +943,15 @@ private void processWorkCompleteMailBox() {
workMailBox.drainTo(results, size);
log.trace("Processing drained work {}...", results.size());
- for (var work : results) {
- MDC.put("offset", work.toString());
- wm.handleFutureResult(work);
- MDC.clear();
+ for (var action : results) {
+ if (action.isNewConsumerRecords()) {
+ wm.registerWork(action.getConsumerRecords());
+ } else {
+ WorkContainer work = action.getWorkContainer();
+ MDC.put(MDC_OFFSET_MARKER, work.toString());
+ wm.handleFutureResult(work);
+ MDC.remove(MDC_OFFSET_MARKER);
+ }
}
}
@@ -926,19 +959,11 @@ private void processWorkCompleteMailBox() {
* The amount of time to block poll in this cycle
*
* @return either the duration until next commit, or next work retry
- * @see WorkManager#isStarvedForNewWork()
- * @see WorkManager#getTotalWorkAwaitingIngestion()
* @see ParallelConsumerOptions#getTargetAmountOfRecordsInFlight()
*/
private Duration getTimeToBlockFor() {
- // should not block as not enough work is being done, and there's more work to ingest
- boolean ingestionWorkAndStarved = wm.hasWorkAwaitingIngestionToShards() && wm.isStarvedForNewWork();
- if (ingestionWorkAndStarved) {
- log.debug("Work waiting to be ingested, and not enough work in flight - will not block");
- return Duration.ofMillis(0);
- }
// if less than target work already in flight, don't sleep longer than the next retry time for failed work, if it exists - so that we can wake up and maybe retry the failed work
- else if (!wm.isWorkInFlightMeetingTarget()) {
+ if (!wm.isWorkInFlightMeetingTarget()) {
// though check if we have work awaiting retry
var lowestScheduledOpt = wm.getLowestRetryTime();
if (lowestScheduledOpt.isPresent()) {
@@ -1015,7 +1040,7 @@ private boolean lingeringOnCommitWouldBeBeneficial() {
workIsWaitingToBeCompletedSuccessfully, workInFlight, workWaitingInMailbox, !workWaitingToCommit);
boolean result = workIsWaitingToBeCompletedSuccessfully || workInFlight || workWaitingInMailbox || !workWaitingToCommit;
- // disable - commit frequency takes care of lingering? is this outdated?
+ // todo disable - commit frequency takes care of lingering? is this outdated?
return false;
}
@@ -1097,7 +1122,12 @@ protected List, R>> runUserFunct
return intermediateResults;
} catch (Exception e) {
// handle fail
- log.error("Exception caught in user function running stage, registering WC as failed, returning to mailbox", e);
+ String msg = "Exception caught in user function running stage, registering WC as failed, returning to mailbox";
+ if (e instanceof RetriableException) {
+ log.debug("Explicit " + RetriableException.class.getSimpleName() + " caught, logging at DEBUG only. " + msg, e);
+ } else {
+ log.error(msg, e);
+ }
for (var wc : workContainerBatch) {
wc.onUserFunctionFailure(e);
addToMailbox(wc); // always add on error
@@ -1117,8 +1147,13 @@ protected void onUserFunctionSuccess(WorkContainer wc, List resultsFrom
protected void addToMailbox(WorkContainer wc) {
String state = wc.isUserFunctionSucceeded() ? "succeeded" : "FAILED";
- log.debug("Adding {} {} to mailbox...", state, wc);
- workMailBox.add(wc);
+ log.trace("Adding {} {} to mailbox...", state, wc);
+ workMailBox.add(ControllerEventMessage.of(wc));
+ }
+
+ public void registerWork(EpochAndRecordsMap polledRecords) {
+ log.debug("Adding {} to mailbox...", polledRecords);
+ workMailBox.add(ControllerEventMessage.of(polledRecords));
}
/**
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/BrokerPollSystem.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/BrokerPollSystem.java
index ea7d3826c..3fc470eb7 100644
--- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/BrokerPollSystem.java
+++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/BrokerPollSystem.java
@@ -59,11 +59,12 @@ public class BrokerPollSystem implements OffsetCommitter {
private final WorkManager wm;
- public BrokerPollSystem(ConsumerManager consumerMgr, WorkManager wm, AbstractParallelEoSStreamProcessor pc, final ParallelConsumerOptions options) {
+ public BrokerPollSystem(ConsumerManager consumerMgr, WorkManager wm, AbstractParallelEoSStreamProcessor pc, final ParallelConsumerOptions options) {
this.wm = wm;
this.pc = pc;
this.consumerManager = consumerMgr;
+
switch (options.getCommitMode()) {
case PERIODIC_CONSUMER_SYNC, PERIODIC_CONSUMER_ASYNCHRONOUS -> {
ConsumerOffsetCommitter consumerCommitter = new ConsumerOffsetCommitter<>(consumerMgr, wm, options);
@@ -107,22 +108,7 @@ private boolean controlLoop() {
committer.ifPresent(x -> x.claim());
try {
while (state != closed) {
- log.trace("Loop: Broker poller: ({})", state);
- if (state == running || state == draining) { // if draining - subs will be paused, so use this to just sleep
- ConsumerRecords polledRecords = pollBrokerForRecords();
- log.debug("Got {} records in poll result", polledRecords.count());
-
- if (!polledRecords.isEmpty()) {
- log.trace("Loop: Register work");
- wm.registerWork(polledRecords);
-
- // notify control work has been registered, in case it's sleeping waiting for work that will never come
- if (wm.isStarvedForNewWork()) {
- log.trace("Apparently no work is being done, make sure Control is awake to receive messages");
- pc.notifySomethingToDo();
- }
- }
- }
+ handlePoll();
maybeDoCommit();
@@ -143,6 +129,20 @@ private boolean controlLoop() {
}
}
+ private void handlePoll() {
+ log.trace("Loop: Broker poller: ({})", state);
+ if (state == running || state == draining) { // if draining - subs will be paused, so use this to just sleep
+ var polledRecords = pollBrokerForRecords();
+ int count = polledRecords.count();
+ log.debug("Got {} records in poll result", count);
+
+ if (count > 0) {
+ log.trace("Loop: Register work");
+ pc.registerWork(polledRecords);
+ }
+ }
+ }
+
private void doClose() {
log.debug("Doing close...");
doPause();
@@ -166,7 +166,7 @@ private boolean isResponsibleForCommits() {
return committer.isPresent();
}
- private ConsumerRecords pollBrokerForRecords() {
+ private EpochAndRecordsMap pollBrokerForRecords() {
managePauseOfSubscription();
log.debug("Subscriptions are paused: {}", paused);
@@ -176,8 +176,11 @@ private ConsumerRecords pollBrokerForRecords() {
log.debug("Long polling broker with timeout {}, might appear to sleep here if subs are paused, or no data available on broker. Run state: {}", thisLongPollTimeout, state);
ConsumerRecords poll = consumerManager.poll(thisLongPollTimeout);
+
log.debug("Poll completed");
- return poll;
+
+ // build records map
+ return new EpochAndRecordsMap<>(poll, wm.getPm());
}
/**
@@ -205,10 +208,9 @@ private void doPauseMaybe() {
});
} else {
if (log.isDebugEnabled()) {
- log.debug("Should pause but pause rate limit exceeded {} vs {}. Queued: {}",
+ log.debug("Should pause but pause rate limit exceeded {} vs {}.",
pauseLimiter.getElapsedDuration(),
- pauseLimiter.getRate(),
- wm.getAmountOfWorkQueuedWaitingIngestion());
+ pauseLimiter.getRate());
}
}
}
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/CountingCRLinkedList.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/CountingCRLinkedList.java
deleted file mode 100644
index 28cbcb201..000000000
--- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/CountingCRLinkedList.java
+++ /dev/null
@@ -1,53 +0,0 @@
-package io.confluent.parallelconsumer.internal;
-
-/*-
- * Copyright (C) 2020-2021 Confluent, Inc.
- */
-
-import lombok.EqualsAndHashCode;
-import lombok.Getter;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-
-import java.util.LinkedList;
-import java.util.Queue;
-
-/**
- * Cached view of the nested number of records in this queue.
- *
- * Also protects against concurrent modification exceptions, as we don't need to traverse the list to count the nested
- * elements. The count won't then be always exact, but it doesn't need to be.
- */
-@EqualsAndHashCode(callSuper = true)
-public class CountingCRLinkedList extends LinkedList> implements Queue> {
-
- /**
- * The number of nested {@link ConsumerRecord} in this collection. As this is a non blocking collection, this won't
- * be exact.
- */
- @Getter
- private int nestedCount = 0;
-
- @Override
- public void add(final int index, final ConsumerRecords element) {
- nestedCount = nestedCount + element.count();
- super.add(index, element);
- }
-
- @Override
- public boolean add(final ConsumerRecords element) {
- nestedCount = nestedCount + element.count();
- return super.add(element);
- }
-
- @Override
- public ConsumerRecords poll() {
- ConsumerRecords poll = super.poll();
- if (poll != null) {
- int numberOfNestedMessages = poll.count();
- nestedCount = nestedCount - numberOfNestedMessages;
- }
- return poll;
- }
-
-}
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/EpochAndRecordsMap.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/EpochAndRecordsMap.java
new file mode 100644
index 000000000..ffd7c0874
--- /dev/null
+++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/EpochAndRecordsMap.java
@@ -0,0 +1,67 @@
+package io.confluent.parallelconsumer.internal;
+
+/*-
+ * Copyright (C) 2020-2022 Confluent, Inc.
+ */
+
+import io.confluent.parallelconsumer.state.PartitionStateManager;
+import lombok.Value;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.*;
+
+/**
+ * For tagging polled records with our epoch
+ *
+ * @see BrokerPollSystem#partitionAssignmentEpoch
+ */
+@Value
+public class EpochAndRecordsMap {
+
+ Map recordMap = new HashMap<>();
+
+ public EpochAndRecordsMap(ConsumerRecords poll, PartitionStateManager pm) {
+ poll.partitions().forEach(partition -> {
+ var records = poll.records(partition);
+ Long epochOfPartition = pm.getEpochOfPartition(partition);
+ recordMap.put(partition, new RecordsAndEpoch(epochOfPartition, records));
+ });
+ }
+
+ /**
+ * Get the partitions which have records contained in this record set.
+ *
+ * @return the set of partitions with data in this record set (may be empty if no data was returned)
+ */
+ public Set partitions() {
+ return Collections.unmodifiableSet(recordMap.keySet());
+ }
+
+ /**
+ * Get just the records for the given partition
+ *
+ * @param partition The partition to get records for
+ */
+ public RecordsAndEpoch records(TopicPartition partition) {
+ return this.recordMap.get(partition);
+ }
+
+ /**
+ * The number of records for all topics
+ */
+ public int count() {
+ int count = 0;
+ for (var recs : this.recordMap.values())
+ count += recs.getRecords().size();
+ return count;
+ }
+
+ @Value
+ public class RecordsAndEpoch {
+ Long epochOfPartitionAtPoll;
+ List> records;
+ }
+
+}
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/BitSetEncoder.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/BitSetEncoder.java
index 1ecdc556e..04dcb05dc 100644
--- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/BitSetEncoder.java
+++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/BitSetEncoder.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.csid.utils.StringUtils;
@@ -108,7 +108,7 @@ private ByteBuffer initV2(int bitsetEntriesRequired) throws BitSetEncodingNotSup
private ByteBuffer initV1(int bitsetEntriesRequired) throws BitSetEncodingNotSupportedException {
if (bitsetEntriesRequired > Short.MAX_VALUE) {
// need to upgrade to using Integer for the bitset length, but can't change serialisation format in-place
- throw new BitSetEncodingNotSupportedException("BitSet V1 too long to encode, bitset length overflows Short.MAX_VALUE: " + bitsetEntriesRequired + ". (max: " + Short.MAX_VALUE + ")");
+ throw new BitSetEncodingNotSupportedException("Input too long to encode for BitSet V1, length overflows Short.MAX_VALUE: " + bitsetEntriesRequired + ". (max: " + Short.MAX_VALUE + ")");
}
// prep bit set buffer
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/EncodedOffsetPair.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/EncodedOffsetPair.java
index 7a9697969..65dd6318e 100644
--- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/EncodedOffsetPair.java
+++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/EncodedOffsetPair.java
@@ -11,6 +11,7 @@
import lombok.extern.slf4j.Slf4j;
import java.nio.ByteBuffer;
+import java.util.Comparator;
import static io.confluent.parallelconsumer.offsets.OffsetBitSet.deserialiseBitSetWrap;
import static io.confluent.parallelconsumer.offsets.OffsetBitSet.deserialiseBitSetWrapToIncompletes;
@@ -27,6 +28,7 @@
@Slf4j
public final class EncodedOffsetPair implements Comparable {
+ public static final Comparator SIZE_COMPARATOR = Comparator.comparingInt(x -> x.data.capacity());
@Getter
OffsetEncoding encoding;
@Getter
@@ -42,7 +44,7 @@ public final class EncodedOffsetPair implements Comparable {
@Override
public int compareTo(EncodedOffsetPair o) {
- return Integer.compare(data.capacity(), o.getData().capacity());
+ return SIZE_COMPARATOR.compare(this, o);
}
/**
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 9054b411b..2149b0231 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
@@ -189,7 +189,12 @@ String serialiseIncompleteOffsetMapToBase64(long finalOffsetForPartition, Partit
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);
+ if (log.isDebugEnabled()) {
+ log.debug("Encoding partition {}, highest succeeded {}, incomplete offsets to encode {}",
+ partitionState.getTp(),
+ highestSucceeded,
+ partitionState.getIncompleteOffsetsBelowHighestSucceeded());
+ }
OffsetSimultaneousEncoder simultaneousEncoder = new OffsetSimultaneousEncoder(finalOffsetForPartition, highestSucceeded, incompleteOffsets).invoke();
//
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 237bccdad..55471af66 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
@@ -61,8 +61,7 @@ public class OffsetSimultaneousEncoder {
* @see #packSmallest()
*/
@Getter
- PriorityQueue sortedEncodings = new PriorityQueue<>();
-
+ SortedSet sortedEncodings = new TreeSet<>();
/**
* Force the encoder to also add the compressed versions. Useful for testing.
@@ -226,7 +225,7 @@ public byte[] packSmallest() throws NoEncodingPossibleException {
if (sortedEncodings.isEmpty()) {
throw new NoEncodingPossibleException("No encodings could be used");
}
- final EncodedOffsetPair best = this.sortedEncodings.poll();
+ final EncodedOffsetPair best = this.sortedEncodings.first();
log.debug("Compression chosen is: {}", best.encoding.name());
return packEncoding(best);
}
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 e2ccf4094..ebf33fe81 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
@@ -9,6 +9,7 @@
import io.confluent.parallelconsumer.offsets.OffsetMapCodecManager;
import lombok.Getter;
import lombok.Setter;
+import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
@@ -27,9 +28,16 @@
import static java.util.Optional.of;
import static lombok.AccessLevel.*;
+@ToString
@Slf4j
public class PartitionState {
+ /**
+ * Symbolic value for a parameter which is initialised as having an offset absent (instead of using Optional or
+ * null)
+ */
+ public static final long KAFKA_OFFSET_ABSENCE = -1L;
+
@Getter
private final TopicPartition tp;
@@ -71,7 +79,7 @@ public class PartitionState {
* Highest offset which has completed successfully ("succeeded").
*/
@Getter(PUBLIC)
- private long offsetHighestSucceeded = -1L;
+ private long offsetHighestSucceeded = KAFKA_OFFSET_ABSENCE;
/**
* If true, more messages are allowed to process for this partition.
@@ -100,6 +108,7 @@ public class PartitionState {
* ({@link #getCommitDataIfDirty()}), or reading upon {@link #onPartitionsRemoved}
*/
// todo doesn't need to be concurrent any more?
+ @ToString.Exclude
private final NavigableMap> commitQueue = new ConcurrentSkipListMap<>();
private NavigableMap> getCommitQueue() {
@@ -108,7 +117,7 @@ private NavigableMap> getCommitQueue() {
public PartitionState(TopicPartition tp, OffsetMapCodecManager.HighestOffsetAndIncompletes offsetData) {
this.tp = tp;
- this.offsetHighestSeen = offsetData.getHighestSeenOffset().orElse(-1L);
+ this.offsetHighestSeen = offsetData.getHighestSeenOffset().orElse(KAFKA_OFFSET_ABSENCE);
this.incompleteOffsets = new ConcurrentSkipListSet<>(offsetData.getIncompleteOffsets());
this.offsetHighestSucceeded = this.offsetHighestSeen;
}
@@ -230,6 +239,7 @@ public Set getAllIncompleteOffsets() {
*/
public Set getIncompleteOffsetsBelowHighestSucceeded() {
long highestSucceeded = getOffsetHighestSucceeded();
+ //noinspection FuseStreamOperations Collectors.toUnmodifiableSet since v10
return Collections.unmodifiableSet(incompleteOffsets.parallelStream()
// todo less than or less than and equal?
.filter(x -> x < highestSucceeded)
@@ -306,10 +316,21 @@ private boolean updateBlockFromEncodingResult(String offsetMapPayload) {
}
private double getPressureThresholdValue() {
- return DefaultMaxMetadataSize * PartitionMonitor.getUSED_PAYLOAD_THRESHOLD_MULTIPLIER();
+ return DefaultMaxMetadataSize * PartitionStateManager.getUSED_PAYLOAD_THRESHOLD_MULTIPLIER();
}
public void onPartitionsRemoved(ShardManager sm) {
sm.removeAnyShardsReferencedBy(getCommitQueue());
}
+
+ /**
+ * Convenience method for readability
+ *
+ * @return true if {@link #isAllowedMoreRecords()} is false
+ * @see #isAllowedMoreRecords()
+ */
+ public boolean isBlocked() {
+ return !isAllowedMoreRecords();
+ }
}
+
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/PartitionStateManager.java
similarity index 76%
rename from parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionMonitor.java
rename to parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionStateManager.java
index 65c19b2dd..1ec416567 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/PartitionStateManager.java
@@ -8,6 +8,7 @@
import io.confluent.parallelconsumer.ParallelConsumerOptions.ProcessingOrder;
import io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor;
import io.confluent.parallelconsumer.internal.BrokerPollSystem;
+import io.confluent.parallelconsumer.internal.EpochAndRecordsMap;
import io.confluent.parallelconsumer.internal.InternalRuntimeError;
import io.confluent.parallelconsumer.offsets.OffsetMapCodecManager;
import lombok.Getter;
@@ -21,10 +22,7 @@
import org.apache.kafka.common.TopicPartition;
import java.time.Clock;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
@@ -35,12 +33,14 @@
* In charge of managing {@link PartitionState}s.
*
* This state is shared between the {@link BrokerPollSystem} thread and the {@link AbstractParallelEoSStreamProcessor}.
+ *
+ * @see PartitionState
*/
@Slf4j
@RequiredArgsConstructor
-// todo rename to partition manager
-public class PartitionMonitor implements ConsumerRebalanceListener {
+public class PartitionStateManager implements ConsumerRebalanceListener {
+ public static final double USED_PAYLOAD_THRESHOLD_MULTIPLIER_DEFAULT = 0.75;
/**
* Best efforts attempt to prevent usage of offset payload beyond X% - as encoding size test is currently only done
* per batch, we need to leave some buffer for the required space to overrun before hitting the hard limit where we
@@ -48,7 +48,8 @@ public class PartitionMonitor implements ConsumerRebalanceListener {
*/
@Getter
@Setter
- private static double USED_PAYLOAD_THRESHOLD_MULTIPLIER = 0.75;
+ // todo remove static
+ private static double USED_PAYLOAD_THRESHOLD_MULTIPLIER = USED_PAYLOAD_THRESHOLD_MULTIPLIER_DEFAULT;
private final Consumer consumer;
@@ -64,20 +65,18 @@ public class PartitionMonitor implements ConsumerRebalanceListener {
/**
* Record the generations of partition assignment, for fencing off invalid work.
*
- * This must live outside of {@link PartitionState}, as it must be tracked across partition lifecycles.
+ * NOTE: This must live outside of {@link PartitionState}, as it must be tracked across partition lifecycles.
*
* Starts at zero.
+ *
+ * NOTE: Must be concurrent because it can be set by one thread, but read by another.
*/
- private final Map partitionsAssignmentEpochs = new ConcurrentHashMap<>();
+ private final Map partitionsAssignmentEpochs = new ConcurrentHashMap<>();
private final Clock clock;
public PartitionState getPartitionState(TopicPartition tp) {
- // may cause the system to wait for a rebalance to finish
- // by locking on partitionState, may cause the system to wait for a rebalance to finish
- synchronized (partitionStates) {
- return partitionStates.get(tp);
- }
+ return partitionStates.get(tp);
}
/**
@@ -86,33 +85,30 @@ public PartitionState getPartitionState(TopicPartition tp) {
@Override
public void onPartitionsAssigned(Collection assignedPartitions) {
log.debug("Partitions assigned: {}", assignedPartitions);
- synchronized (this.partitionStates) {
-
- 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: {}", partitionAssignment, previouslyAssignedState);
- }
+
+ 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: {}", partitionAssignment, previouslyAssignedState);
}
}
+ }
- incrementPartitionAssignmentEpoch(assignedPartitions);
-
- try {
- 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);
- throw e;
- }
+ incrementPartitionAssignmentEpoch(assignedPartitions);
+ try {
+ 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);
+ throw e;
}
}
@@ -136,10 +132,8 @@ public void onPartitionsRevoked(Collection partitions) {
}
void onPartitionsRemoved(final Collection partitions) {
- synchronized (this.partitionStates) {
- incrementPartitionAssignmentEpoch(partitions);
- resetOffsetMapAndRemoveWork(partitions);
- }
+ incrementPartitionAssignmentEpoch(partitions);
+ resetOffsetMapAndRemoveWork(partitions);
}
/**
@@ -198,19 +192,28 @@ private void resetOffsetMapAndRemoveWork(Collection allRemovedPa
}
}
- public int getEpoch(final ConsumerRecord rec) {
+ /**
+ * @return the current epoch of the partition this record belongs to
+ */
+ public Long getEpochOfPartitionForRecord(final ConsumerRecord rec) {
var tp = toTopicPartition(rec);
- Integer epoch = partitionsAssignmentEpochs.get(tp);
- rec.topic();
+ Long epoch = partitionsAssignmentEpochs.get(tp);
if (epoch == null) {
throw new InternalRuntimeError(msg("Received message for a partition which is not assigned: {}", rec));
}
return epoch;
}
+ /**
+ * @return the current epoch of the partition
+ */
+ public Long getEpochOfPartition(TopicPartition partition) {
+ return partitionsAssignmentEpochs.get(partition);
+ }
+
private void incrementPartitionAssignmentEpoch(final Collection partitions) {
for (final TopicPartition partition : partitions) {
- int epoch = partitionsAssignmentEpochs.getOrDefault(partition, -1);
+ Long epoch = partitionsAssignmentEpochs.getOrDefault(partition, PartitionState.KAFKA_OFFSET_ABSENCE);
epoch++;
partitionsAssignmentEpochs.put(partition, epoch);
}
@@ -227,8 +230,8 @@ private void incrementPartitionAssignmentEpoch(final Collection
boolean checkIfWorkIsStale(final WorkContainer workContainer) {
var topicPartitionKey = workContainer.getTopicPartition();
- Integer currentPartitionEpoch = partitionsAssignmentEpochs.get(topicPartitionKey);
- int workEpoch = workContainer.getEpoch();
+ Long currentPartitionEpoch = partitionsAssignmentEpochs.get(topicPartitionKey);
+ long workEpoch = workContainer.getEpoch();
boolean partitionNotAssigned = isPartitionRemovedOrNeverAssigned(workContainer.getCr());
@@ -327,36 +330,40 @@ public void onFailure(WorkContainer wc) {
/**
* Takes a record as work and puts it into internal queues, unless it's been previously recorded as completed as per
* loaded records.
- *
- * Locking on partition state here, means that the check for assignment is in the same sync block as registering the
- * work with the {@link TopicPartition}'s {@link PartitionState} and the {@link ShardManager}. Keeping the two
- * different views in sync. Of course now, having a shared nothing architecture would mean all access to the state
- * is by a single thread, and so this could never occur (see ).
- *
- * @return true if the record was taken, false if it was skipped (previously successful)
*/
- boolean maybeRegisterNewRecordAsWork(final ConsumerRecord rec) {
- if (rec == null) return false;
+ void maybeRegisterNewRecordAsWork(final EpochAndRecordsMap recordsMap) {
+ for (var partition : recordsMap.partitions()) {
+ var recordsList = recordsMap.records(partition);
+ var epochOfInboundRecords = recordsList.getEpochOfPartitionAtPoll();
+ for (var rec : recordsList.getRecords()) {
+ maybeRegisterNewRecordAsWork(epochOfInboundRecords, rec);
+ }
+ }
+ }
+
+ /**
+ * @see #maybeRegisterNewRecordAsWork(EpochAndRecordsMap)
+ */
+ private void maybeRegisterNewRecordAsWork(Long epochOfInboundRecords, ConsumerRecord rec) {
+ // do epochs still match? do a proactive check, but the epoch will be checked again at work completion as well
+ var currentPartitionEpoch = getEpochOfPartitionForRecord(rec);
+ if (Objects.equals(epochOfInboundRecords, currentPartitionEpoch)) {
- synchronized (partitionStates) {
if (isPartitionRemovedOrNeverAssigned(rec)) {
log.debug("Record in buffer for a partition no longer assigned. Dropping. TP: {} rec: {}", toTopicPartition(rec), rec);
- return false;
}
if (isRecordPreviouslyCompleted(rec)) {
log.trace("Record previously completed, skipping. offset: {}", rec.offset());
- return false;
} else {
- int currentPartitionEpoch = getEpoch(rec);
- var wc = new WorkContainer<>(currentPartitionEpoch, rec, options.getRetryDelayProvider(), clock);
-
- sm.addWorkContainer(wc);
+ var work = new WorkContainer<>(epochOfInboundRecords, rec, options.getRetryDelayProvider(), clock);
- addWorkContainer(wc);
-
- return true;
+ sm.addWorkContainer(work);
+ addWorkContainer(work);
}
+ } else {
+ log.debug("Inbound record of work has epoch ({}) not matching currently assigned epoch for the applicable partition ({}), skipping",
+ epochOfInboundRecords, currentPartitionEpoch);
}
}
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 22ff310fd..9727fcb73 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
@@ -35,6 +35,7 @@ public class ProcessingShard {
* Uses a ConcurrentSkipListMap instead of a TreeMap as under high pressure there appears to be some concurrency
* errors (missing WorkContainers).
*/
+ @Getter
private final NavigableMap> entries = new ConcurrentSkipListMap<>();
@Getter(PRIVATE)
@@ -42,7 +43,7 @@ public class ProcessingShard {
private final ParallelConsumerOptions options;
- private final PartitionMonitor pm;
+ private final PartitionStateManager pm;
private final RateLimiter slowWarningRateLimit = new RateLimiter(5);
@@ -74,9 +75,19 @@ public Optional> getWorkForOffset(long offset) {
}
public long getCountOfWorkAwaitingSelection() {
- return entries.values().parallelStream()
+ return entries.values().stream()
// todo missing pm.isBlocked(topicPartition) ?
- .filter(kvWorkContainer -> kvWorkContainer.isAvailableToTakeAsWork())
+ .filter(WorkContainer::isAvailableToTakeAsWork)
+ .count();
+ }
+
+ public long getCountOfWorkTracked() {
+ return entries.size();
+ }
+
+ public long getCountWorkInFlight() {
+ return entries.values().stream()
+ .filter(WorkContainer::isInFlight)
.count();
}
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 50b2f9749..aef4946f1 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
@@ -18,13 +18,13 @@
* references or {@link Optional}s. By replacing with a no op implementation, we protect for stale messages still in
* queues which reference it, among other things.
*
- * The alternative to this implementation, is having {@link PartitionMonitor#getPartitionState(TopicPartition)} return
- * {@link Optional}, which forces the implicit null check everywhere partition state is retrieved. This was drafted to a
- * degree, but found to be extremely invasive, where this solution with decent separation of concerns and encapsulation,
- * is sufficient and potentially more useful as is non-destructive. Potential issue is that of memory leak as the
- * collection will forever expand. However, even massive partition counts to a single consumer would be in the hundreds
- * of thousands, this would only result in hundreds of thousands of {@link TopicPartition} object keys all pointing to
- * the same instance of {@link RemovedPartitionState}.
+ * The alternative to this implementation, is having {@link PartitionStateManager#getPartitionState(TopicPartition)}
+ * return {@link Optional}, which forces the implicit null check everywhere partition state is retrieved. This was
+ * drafted to a degree, but found to be extremely invasive, where this solution with decent separation of concerns and
+ * encapsulation, is sufficient and potentially more useful as is non-destructive. Potential issue is that of memory
+ * leak as the collection will forever expand. However, even massive partition counts to a single consumer would be in
+ * the hundreds of thousands, this would only result in hundreds of thousands of {@link TopicPartition} object keys all
+ * pointing to the same instance of {@link RemovedPartitionState}.
*/
@Slf4j
public class RemovedPartitionState extends PartitionState {
@@ -82,13 +82,13 @@ public Set getIncompleteOffsetsBelowHighestSucceeded() {
@Override
public long getOffsetHighestSeen() {
log.debug(NO_OP);
- return -1L;
+ return PartitionState.KAFKA_OFFSET_ABSENCE;
}
@Override
public long getOffsetHighestSucceeded() {
log.debug(NO_OP);
- return -1L;
+ return PartitionState.KAFKA_OFFSET_ABSENCE;
}
@Override
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ShardManager.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ShardManager.java
index e8da53748..215494c6e 100644
--- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ShardManager.java
+++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ShardManager.java
@@ -57,7 +57,7 @@ public class ShardManager {
* @see K
* @see WorkManager#getWorkIfAvailable()
*/
- // todo performance: disable/remove if using partition order
+ // performance: could disable/remove if using partition order - but probably not worth the added complexity in the code to handle an extra special case
private final Map> processingShards = new ConcurrentHashMap<>();
private final NavigableSet> retryQueue = new TreeSet<>(Comparator.comparing(wc -> wc.getDelayUntilRetryDue()));
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkContainer.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkContainer.java
index 584759ac7..a14ed5285 100644
--- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkContainer.java
+++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkContainer.java
@@ -38,7 +38,7 @@ public class WorkContainer implements Comparable> {
* in the system of in flight.
*/
@Getter
- private final int epoch;
+ private final long epoch;
/**
* Simple way to differentiate treatment based on type
@@ -85,7 +85,7 @@ public class WorkContainer implements Comparable> {
// static instance so can't access generics - but don't need them as Options class ensures type is correct
private static Function retryDelayProvider;
- public WorkContainer(int epoch, ConsumerRecord cr, Function, Duration> retryDelayProvider, String workType, Clock clock) {
+ public WorkContainer(long epoch, ConsumerRecord cr, Function, Duration> retryDelayProvider, String workType, Clock clock) {
Objects.requireNonNull(workType);
this.epoch = epoch;
@@ -99,7 +99,7 @@ public WorkContainer(int epoch, ConsumerRecord cr, Function cr, Function, Duration> retryDelayProvider, Clock clock) {
+ public WorkContainer(long epoch, ConsumerRecord cr, Function, Duration> retryDelayProvider, Clock clock) {
this(epoch, cr, retryDelayProvider, DEFAULT_TYPE, clock);
}
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkMailBoxManager.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkMailBoxManager.java
deleted file mode 100644
index 8e9e04a9c..000000000
--- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkMailBoxManager.java
+++ /dev/null
@@ -1,136 +0,0 @@
-package io.confluent.parallelconsumer.state;
-
-/*-
- * Copyright (C) 2020-2022 Confluent, Inc.
- */
-import io.confluent.parallelconsumer.internal.BrokerPollSystem;
-import io.confluent.parallelconsumer.internal.CountingCRLinkedList;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.common.TopicPartition;
-
-import java.util.Collection;
-import java.util.LinkedList;
-import java.util.Queue;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import static io.confluent.csid.utils.KafkaUtils.toTopicPartition;
-
-/**
- * Handles the incoming mail for {@link WorkManager}.
- */
-@Slf4j
-public class WorkMailBoxManager {
-
- /**
- * The number of nested {@link ConsumerRecord} entries in the shared blocking mail box. Cached for performance.
- */
- private int sharedBoxNestedRecordCount;
-
- /**
- * The shared mailbox. Doesn't need to be thread safe as we already need synchronize on it.
- */
- private final LinkedBlockingQueue> workInbox = new LinkedBlockingQueue<>();
-
- /**
- * Mailbox where mail is transferred to immediately.
- */
- private final CountingCRLinkedList internalBatchMailQueue = new CountingCRLinkedList<>();
-
- /**
- * Queue of records flattened from the {@link #internalBatchMailQueue}.
- *
- * This is needed because {@link java.util.concurrent.BlockingQueue#drainTo(Collection)} must drain to a collection
- * of the same type. We could have {@link BrokerPollSystem} do the flattening, but that would require many calls to
- * the Concurrent queue, where this only needs one. Also as we don't expect there to be that many elements in these
- * collections (as they contain large batches of records), the overhead will be small.
- */
- private final Queue> internalFlattenedMailQueue = new LinkedList<>();
-
- /**
- * @return amount of work queued in the mail box, awaiting processing into shards, not exact
- */
- Integer getAmountOfWorkQueuedWaitingIngestion() {
- return sharedBoxNestedRecordCount +
- internalBatchMailQueue.getNestedCount() +
- internalFlattenedMailQueue.size();
- }
-
- /**
- * Work must be registered in offset order
- *
- * Thread safe for use by control and broker poller thread.
- *
- * @see WorkManager#onSuccessResult
- * @see WorkManager#raisePartitionHighWaterMark
- */
- public void registerWork(final ConsumerRecords records) {
- synchronized (workInbox) {
- sharedBoxNestedRecordCount += records.count();
- workInbox.add(records);
- }
- }
-
-
- /**
- * Must synchronise to keep sharedBoxNestedRecordCount in lock step with the inbox. Register is easy, but drain you
- * need to run through an intermediary collection and then count the nested elements, to know how many to subtract
- * from the Atomic nested count.
- *
- * Plus registering work is relatively infrequent, so shouldn't worry about a little synchronized here - makes it
- * much simpler.
- */
- private void drainSharedMailbox() {
- synchronized (workInbox) {
- workInbox.drainTo(internalBatchMailQueue);
- sharedBoxNestedRecordCount = 0;
- }
- }
-
- /**
- * Take our inbound messages from the {@link BrokerPollSystem} and add them to our registry.
- */
- private synchronized void flattenBatchQueue() {
- drainSharedMailbox();
-
- // flatten
- while (!internalBatchMailQueue.isEmpty()) {
- ConsumerRecords consumerRecords = internalBatchMailQueue.poll();
- log.debug("Flattening {} records", consumerRecords.count());
- for (final ConsumerRecord consumerRecord : consumerRecords) {
- internalFlattenedMailQueue.add(consumerRecord);
- }
- }
- }
-
- /**
- * Remove revoked work from the mailbox
- */
- public synchronized void onPartitionsRemoved(final Collection removedPartitions) {
- log.debug("Removing stale work from inbox queues");
- flattenBatchQueue();
- internalFlattenedMailQueue.removeIf(rec ->
- removedPartitions.contains(toTopicPartition(rec))
- );
- }
-
- public synchronized boolean internalFlattenedMailQueueIsEmpty() {
- return internalFlattenedMailQueue.isEmpty();
- }
-
- /**
- * @return the next element in our outbound queue, or null if empty
- */
- public synchronized ConsumerRecord internalFlattenedMailQueuePoll() {
- if (internalBatchMailQueue.isEmpty()) {
- // flatten the batch queue in batches when needed
- flattenBatchQueue();
- }
- return internalFlattenedMailQueue.poll();
- }
-
- public int internalFlattenedMailQueueSize() {
- return internalFlattenedMailQueue.size();
- }
-}
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 93d3aecd1..d92317405 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
@@ -9,11 +9,10 @@
import io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor;
import io.confluent.parallelconsumer.internal.BrokerPollSystem;
import io.confluent.parallelconsumer.internal.DynamicLoadFactor;
+import io.confluent.parallelconsumer.internal.EpochAndRecordsMap;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import pl.tlinkowski.unij.api.UniLists;
@@ -45,12 +44,11 @@
public class WorkManager implements ConsumerRebalanceListener {
@Getter
- private final ParallelConsumerOptions options;
+ private final ParallelConsumerOptions options;
- // todo rename PSM, PartitionStateManager
// todo make private
@Getter(PUBLIC)
- final PartitionMonitor pm;
+ final PartitionStateManager pm;
// todo make private
@Getter(PUBLIC)
@@ -64,8 +62,6 @@ public class WorkManager implements ConsumerRebalanceListener {
*/
private final DynamicLoadFactor dynamicLoadFactor;
- private final WorkMailBoxManager wmbm;
-
@Getter
private int numberRecordsOutForProcessing = 0;
@@ -90,9 +86,8 @@ public WorkManager(final ParallelConsumerOptions newOptions, final org.apa
final DynamicLoadFactor dynamicExtraLoadFactor, Clock clock) {
this.options = newOptions;
this.dynamicLoadFactor = dynamicExtraLoadFactor;
- this.wmbm = new WorkMailBoxManager<>();
this.sm = new ShardManager<>(options, this, clock);
- this.pm = new PartitionMonitor<>(consumer, sm, options, clock);
+ this.pm = new PartitionStateManager<>(consumer, sm, options, clock);
}
/**
@@ -126,40 +121,11 @@ public void onPartitionsLost(Collection partitions) {
}
void onPartitionsRemoved(final Collection partitions) {
- wmbm.onPartitionsRemoved(partitions);
+ // no-op - nothing to do
}
- public void registerWork(ConsumerRecords records) {
- wmbm.registerWork(records);
- }
-
- /**
- * Moves the requested amount of work from initial queues into work queues, if available.
- *
- * @param requestedMaxWorkToRetrieve try to move at least this many messages into the inbound queues
- * @return the number of extra records ingested due to request
- */
- private int ingestPolledRecordsIntoQueues(long requestedMaxWorkToRetrieve) {
- log.debug("Will attempt to register the requested {} - {} available in internal mailbox",
- requestedMaxWorkToRetrieve, wmbm.internalFlattenedMailQueueSize());
-
- //
- var takenWorkCount = 0;
- boolean continueIngesting;
- do {
- ConsumerRecord polledRecord = wmbm.internalFlattenedMailQueuePoll();
- boolean recordAddedAsWork = pm.maybeRegisterNewRecordAsWork(polledRecord);
- if (recordAddedAsWork) {
- takenWorkCount++;
- }
- boolean polledQueueNotExhausted = polledRecord != null;
- boolean ingestTargetNotSatisfied = takenWorkCount < requestedMaxWorkToRetrieve;
- continueIngesting = ingestTargetNotSatisfied && polledQueueNotExhausted;
- } while (continueIngesting);
-
- log.debug("{} new records were registered.", takenWorkCount);
-
- return takenWorkCount;
+ public void registerWork(EpochAndRecordsMap records) {
+ pm.maybeRegisterNewRecordAsWork(records);
}
/**
@@ -178,40 +144,20 @@ public List> getWorkIfAvailable(final int requestedMaxWorkTo
return UniLists.of();
}
- int ingested = tryToEnsureQuantityOfWorkQueuedAvailable(requestedMaxWorkToRetrieve);
-
//
var work = sm.getWorkIfAvailable(requestedMaxWorkToRetrieve);
//
- log.debug("Got {} records of work. In-flight: {}, Awaiting in commit (partition) queues: {}", work.size(), getNumberRecordsOutForProcessing(), getNumberOfEntriesInPartitionQueues());
+ log.debug("Got {} of {} requested records of work. In-flight: {}, Awaiting in commit (partition) queues: {}",
+ work.size(),
+ requestedMaxWorkToRetrieve,
+ getNumberRecordsOutForProcessing(),
+ getNumberOfEntriesInPartitionQueues());
numberRecordsOutForProcessing += work.size();
return work;
}
- /**
- * Tries to ensure there are at least this many records available in the queues
- *
- * @return the number of extra records ingested due to request
- */
- // todo rename - shunt messages from internal buffer into queues
- // 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;
- log.debug("Requested: {}, available in shards: {}, will try to process from mailbox the delta of: {}",
- requestedMaxWorkToRetrieve, available, extraNeededFromInboxToSatisfy);
-
- int ingested = ingestPolledRecordsIntoQueues(extraNeededFromInboxToSatisfy);
- log.debug("Ingested an extra {} records", ingested);
-
- long ingestionOffBy = extraNeededFromInboxToSatisfy - ingested;
-
- return ingested;
- }
-
public void onSuccessResult(WorkContainer wc) {
log.trace("Work success ({}), removing from processing shard queue", wc);
@@ -230,7 +176,7 @@ public void onSuccessResult(WorkContainer wc) {
/**
* Can run from controller or poller thread, depending on which is responsible for committing
*
- * @see PartitionMonitor#onOffsetCommitSuccess(Map)
+ * @see PartitionStateManager#onOffsetCommitSuccess(Map)
*/
public void onOffsetCommitSuccess(Map committed) {
pm.onOffsetCommitSuccess(committed);
@@ -248,10 +194,6 @@ public long getNumberOfEntriesInPartitionQueues() {
return pm.getNumberOfEntriesInPartitionQueues();
}
- public Integer getAmountOfWorkQueuedWaitingIngestion() {
- return wmbm.getAmountOfWorkQueuedWaitingIngestion();
- }
-
public Map collectCommitDataForDirtyPartitions() {
return pm.collectDirtyCommitData();
}
@@ -287,7 +229,7 @@ public boolean shouldThrottle() {
* should be downloaded (or pipelined in the Consumer)
*/
public boolean isSufficientlyLoaded() {
- return getAmountOfWorkQueuedWaitingIngestion() > options.getTargetAmountOfRecordsInFlight() * getLoadingFactor();
+ return getNumberOfWorkQueuedInShardsAwaitingSelection() > (long) options.getTargetAmountOfRecordsInFlight() * getLoadingFactor();
}
private int getLoadingFactor() {
@@ -306,17 +248,8 @@ public boolean isWorkInFlightMeetingTarget() {
return getNumberRecordsOutForProcessing() >= options.getTargetAmountOfRecordsInFlight();
}
- /**
- * @return Work count in mailbox plus work added to the processing shards
- */
- public long getTotalWorkAwaitingIngestion() {
- long workQueuedInShardsCount = sm.getNumberOfWorkQueuedInShardsAwaitingSelection();
- Integer workQueuedInMailboxCount = getAmountOfWorkQueuedWaitingIngestion();
- return workQueuedInShardsCount + workQueuedInMailboxCount;
- }
-
- public boolean hasWorkAwaitingIngestionToShards() {
- return getAmountOfWorkQueuedWaitingIngestion() > 0;
+ public long getNumberOfWorkQueuedInShardsAwaitingSelection() {
+ return sm.getNumberOfWorkQueuedInShardsAwaitingSelection();
}
public boolean hasWorkInCommitQueues() {
@@ -324,9 +257,7 @@ public boolean hasWorkInCommitQueues() {
}
public boolean isRecordsAwaitingProcessing() {
- long partitionWorkRemainingCount = sm.getNumberOfWorkQueuedInShardsAwaitingSelection();
- boolean internalQueuesNotEmpty = hasWorkAwaitingIngestionToShards();
- return partitionWorkRemainingCount > 0 || internalQueuesNotEmpty;
+ return sm.getNumberOfWorkQueuedInShardsAwaitingSelection() > 0;
}
public boolean isRecordsAwaitingToBeCommitted() {
@@ -361,12 +292,4 @@ public Optional getLowestRetryTime() {
return sm.getLowestRetryTime();
}
- /**
- * @return true if more records are needed to be sent out for processing (not enough in queues to satisfy
- * concurrency target)
- */
- public boolean isStarvedForNewWork() {
- long queued = getTotalWorkAwaitingIngestion();
- return queued < options.getTargetAmountOfRecordsInFlight();
- }
}
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 f12d7f7b3..402c51536 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
@@ -74,7 +74,8 @@ void load(CommitMode commitMode) {
//
allMessagesConsumedLatch.await(defaultTimeoutSeconds, SECONDS);
// waitAtMost(defaultTimeout).until(() -> producerSpy.consumerGroupOffsetsHistory().size() > 0);
- parallelConsumer.waitForProcessedNotCommitted(defaultTimeout.multipliedBy(10));
+
+
parallelConsumer.close();
// assert quantity of produced messages
@@ -287,7 +288,7 @@ private void testTiming(int numberOfKeys, int quantityOfMessagesToProduce) {
throw new RuntimeException("bad step, expected message(s) is missing: " + missing);
}
- assertThat(producerSpy.history().size()).as("Finally, all messages expected messages were produced").isEqualTo(quantityOfMessagesToProduce);
+ assertThat(producerSpy.history()).as("Finally, all messages expected messages were produced").hasSize(quantityOfMessagesToProduce);
if (isUsingTransactionalProducer()) {
List>> groupOffsetsHistory = producerSpy.consumerGroupOffsetsHistory(); // tx
assertThat(groupOffsetsHistory).as("No offsets committed").hasSizeGreaterThan(0); // tx
diff --git a/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/MultiInstanceHighVolumeTest.java b/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/MultiInstanceHighVolumeTest.java
index a0876227a..035541042 100644
--- a/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/MultiInstanceHighVolumeTest.java
+++ b/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/MultiInstanceHighVolumeTest.java
@@ -43,7 +43,7 @@ class MultiInstanceHighVolumeTest extends BrokerIntegrationTest
public AtomicInteger processedCount = new AtomicInteger(0);
public AtomicInteger producedCount = new AtomicInteger(0);
- int maxPoll = 500;
+ int maxPoll = 500; // 500 is the kafka default
ParallelConsumerOptions.CommitMode commitMode = ParallelConsumerOptions.CommitMode.PERIODIC_CONSUMER_SYNC;
ParallelConsumerOptions.ProcessingOrder order = ParallelConsumerOptions.ProcessingOrder.KEY;
diff --git a/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/MultiInstanceRebalanceTest.java b/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/MultiInstanceRebalanceTest.java
index 0d0405143..63a66a82f 100644
--- a/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/MultiInstanceRebalanceTest.java
+++ b/parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/MultiInstanceRebalanceTest.java
@@ -291,7 +291,7 @@ public void run() {
Collection duplicates = toCollection(StandardComparisonStrategy.instance()
.duplicatesFrom(getAllConsumedKeys(parallelConsumerRunnablesArray)));
log.info("Duplicate consumed keys (at least one is expected due to the rebalance): {}", duplicates);
- double percentageDuplicateTolerance = 0.1;
+ double percentageDuplicateTolerance = 0.2;
assertThat(duplicates)
.as("There should be few duplicate keys")
.hasSizeLessThan((int) (expectedMessageCount * percentageDuplicateTolerance)); // in some env, there are a lot more. i.e. Jenkins running parallel suits
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 98c8873ed..7ebeda8f1 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
@@ -170,7 +170,8 @@ public static void assertLastCommitIs(final LongPollingMockConsumer> generateRecords(int quantity) {
HashMap>> integerListHashMap = generateRecords(defaultKeys, quantity);
- return flatten(integerListHashMap.values());
+ Collection>> values = integerListHashMap.values();
+ return flatten(values);
}
/**
@@ -210,12 +211,12 @@ public ArrayList> generateRecordsForKey(Integer k
return records;
}
- public List flatten(Collection> listlist) {
- List all = new ArrayList<>();
- for (Collection value : listlist) {
+ public List> flatten(Collection>> listlist) {
+ SortedSet> all = new TreeSet<>(Comparator.comparing(ConsumerRecord::offset));
+ for (Collection> value : listlist) {
all.addAll(value);
}
- return all;
+ return new ArrayList<>(all);
}
@Setter
@@ -246,9 +247,10 @@ public void send(MockConsumer consumerSpy, HashMap consumerSpy, List> records) {
+ log.debug("Sending {} more messages to the consumer stub", records.size());
// send records in `correct` offset order as declared by the input data, regardless of the order of the input list
- List> sorted = new ArrayList(records);
- Collections.sort(sorted, Comparator.comparingLong(ConsumerRecord::offset));
+ List> sorted = new ArrayList<>(records);
+ sorted.sort(Comparator.comparingLong(ConsumerRecord::offset));
for (ConsumerRecord record : sorted) {
consumerSpy.addRecord(record);
}
@@ -275,7 +277,7 @@ public static void completeWork(final WorkManager wmm, final Wor
}
public List> sendRecords(final int i) {
- List> consumerRecords = generateRecords(i);
+ var consumerRecords = generateRecords(i);
send(consumerSpy, consumerRecords);
return consumerRecords;
}
diff --git a/parallel-consumer-core/src/test/java/io/confluent/csid/utils/LatchTestUtils.java b/parallel-consumer-core/src/test/java/io/confluent/csid/utils/LatchTestUtils.java
index 408c89ea6..888e8f644 100644
--- a/parallel-consumer-core/src/test/java/io/confluent/csid/utils/LatchTestUtils.java
+++ b/parallel-consumer-core/src/test/java/io/confluent/csid/utils/LatchTestUtils.java
@@ -51,7 +51,7 @@ public static void awaitLatch(final CountDownLatch latch, final int seconds) {
seconds, seconds - toSeconds(between(start, now())));
}
if (latchReachedZero) {
- log.trace("Latch released");
+ log.trace("Latch was released (#countdown)");
} else {
throw new TimeoutException("Latch await timeout (" + seconds + " seconds) - " + latch.getCount() + " count remaining");
}
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 2988f5540..6733bbcc1 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
@@ -46,6 +46,7 @@
import static org.mockito.Mockito.*;
import static pl.tlinkowski.unij.api.UniLists.of;
+// todo migrate commit assertion methods in to a Truth Subject
@Slf4j
public abstract class AbstractParallelEoSStreamProcessorTestBase {
@@ -143,7 +144,7 @@ public void close() {
if (parentParallelConsumer.getFailureCause() != null) {
log.error("PC has error - test failed");
}
- log.debug("Test finished, closing pc...");
+ log.debug("Test ended (maybe a failure), closing pc...");
parentParallelConsumer.close();
} else {
log.debug("Test finished, pc already closed.");
@@ -287,7 +288,7 @@ private void blockingLoopLatchTrigger(int waitForCount) {
loopLatchV = new CountDownLatch(waitForCount);
try {
boolean timeout = !loopLatchV.await(defaultTimeoutSeconds, SECONDS);
- if (timeout)
+ if (timeout || parentParallelConsumer.isClosedOrFailed())
throw new TimeoutException(msg("Timeout of {}, waiting for {} counts, on latch with {} left", defaultTimeout, waitForCount, loopLatchV.getCount()));
} catch (InterruptedException e) {
log.error("Interrupted while waiting for loop latch - timeout was {}", defaultTimeout);
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 1214a9c78..3a8bc97ed 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
@@ -119,7 +119,7 @@ void offsetsAreNeverCommittedForMessagesStillInFlightSimplest(CommitMode commitM
awaitLatch(startBarrierLatch);
// zero records waiting, 2 out for processing
- assertThat(parallelConsumer.getWm().getTotalWorkAwaitingIngestion()).isZero();
+ assertThat(parallelConsumer.getWm().getNumberOfWorkQueuedInShardsAwaitingSelection()).isZero();
assertThat(parallelConsumer.getWm().getNumberRecordsOutForProcessing()).isEqualTo(2);
// finish processing 1
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 d8b8c5b03..1bfdd4703 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
@@ -9,12 +9,12 @@
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 io.confluent.parallelconsumer.state.*;
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 org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.parallel.Isolated;
import org.junit.jupiter.api.parallel.ResourceAccessMode;
@@ -26,6 +26,7 @@
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
@@ -33,6 +34,8 @@
import static io.confluent.csid.utils.JavaUtils.getOnlyOne;
import static io.confluent.csid.utils.LatchTestUtils.awaitLatch;
import static io.confluent.csid.utils.ThreadUtils.sleepQuietly;
+import static io.confluent.parallelconsumer.ManagedTruth.assertWithMessage;
+import static io.confluent.parallelconsumer.state.PartitionStateManager.USED_PAYLOAD_THRESHOLD_MULTIPLIER_DEFAULT;
import static java.time.Duration.ofMillis;
import static java.time.Duration.ofSeconds;
import static java.util.concurrent.TimeUnit.SECONDS;
@@ -48,11 +51,19 @@
* indirectly on the behaviour of the metadata size, even if not so explicitly.
*
* See {@link OffsetMapCodecManager#METADATA_DATA_SIZE_RESOURCE_LOCK}
+ *
+ * @see OffsetMapCodecManager#METADATA_DATA_SIZE_RESOURCE_LOCK
+ * @see OffsetEncodingBackPressureUnitTest
*/
@Isolated // messes with static state - breaks other tests running in parallel
@Slf4j
class OffsetEncodingBackPressureTest extends ParallelEoSStreamProcessorTestBase {
+ @AfterAll
+ static void cleanup() {
+ PartitionStateManager.setUSED_PAYLOAD_THRESHOLD_MULTIPLIER(USED_PAYLOAD_THRESHOLD_MULTIPLIER_DEFAULT);
+ }
+
/**
* Tests that when required space for encoding offset becomes too large, back pressure is put into the system so
* that no further messages for the given partitions can be taken for processing, until more messages complete.
@@ -74,41 +85,55 @@ void backPressureShouldPreventTooManyMessagesBeingQueuedForProcessing() throws O
OffsetMapCodecManager.DefaultMaxMetadataSize = 40; // reduce available to make testing easier
OffsetMapCodecManager.forcedCodec = Optional.of(OffsetEncoding.BitSetV2); // force one that takes a predictable large amount of space
- ktu.send(consumerSpy, ktu.generateRecords(numberOfRecords));
+ //
+ List> records = ktu.generateRecords(numberOfRecords);
+ ktu.send(consumerSpy, records);
+
+ AtomicInteger userFuncFinishedCount = new AtomicInteger();
+ AtomicInteger userFuncStartCount = new AtomicInteger();
- AtomicInteger userFuncFinishedCount = new AtomicInteger(0);
CountDownLatch msgLock = new CountDownLatch(1);
CountDownLatch msgLockTwo = new CountDownLatch(1);
CountDownLatch msgLockThree = new CountDownLatch(1);
- final int numberOfBlockedMessages = 2;
AtomicInteger attempts = new AtomicInteger(0);
long offsetToBlock = 0;
List blockedOffsets = UniLists.of(0L, 2L);
+ final int numberOfBlockedMessages = blockedOffsets.size();
- parallelConsumer.poll((rec) -> {
+ WorkManager wm = parallelConsumer.getWm();
+ final PartitionState partitionState = wm.getPm().getPartitionState(topicPartition);
+ ConcurrentLinkedQueue seen = new ConcurrentLinkedQueue<>();
+
+ parallelConsumer.poll((rec) -> {
+ seen.add(rec.offset());
+ userFuncStartCount.incrementAndGet();
// block the partition to create bigger and bigger offset encoding blocks
// don't let offset 0 finish
if (rec.offset() == offsetToBlock) {
int attemptNumber = attempts.incrementAndGet();
if (attemptNumber == 1) {
- log.debug("force first message to 'never' complete, causing a large offset encoding (lots of messages completing above the low water mark");
- awaitLatch(msgLock, 120);
- log.debug("very slow message awoken, throwing exception");
+ log.debug("Force first message to 'never' complete, causing a large offset encoding (lots of messages completing above the low water mark. Waiting for msgLock countdown.");
+ int timeout = 120;
+ awaitLatch(msgLock, timeout);
+ log.debug("Very slow message awoken, throwing exception");
throw new FakeRuntimeError("Fake error");
} else {
- log.debug("Second attempt, sleeping");
+ log.debug("Second attempt, waiting for msgLockTwo countdown");
awaitLatch(msgLockTwo, 60);
log.debug("Second attempt, unlocked, succeeding");
}
- } else if (rec.offset() == 2l) {
+ } else if (rec.offset() == 2L) {
awaitLatch(msgLockThree);
+ log.debug("// msg 2L unblocked");
} else {
sleepQuietly(1);
}
- userFuncFinishedCount.getAndIncrement();
+ userFuncFinishedCount.incrementAndGet();
});
+ ShardManager sm = wm.getSm();
+
try {
// wait for all pre-produced messages to be processed and produced
@@ -148,7 +173,6 @@ void backPressureShouldPreventTooManyMessagesBeingQueuedForProcessing() throws O
assertThat(highestSeenOffset).as("offset 99 is encoded as having been seen").isEqualTo(expectedHighestSeenOffset);
}
- WorkManager wm = parallelConsumer.getWm();
// partition not blocked
{
@@ -156,7 +180,8 @@ void backPressureShouldPreventTooManyMessagesBeingQueuedForProcessing() throws O
assertThat(partitionBlocked).isFalse();
}
- // feed more messages in order to threshold block - as Bitset requires linearly as much space as we are feeding messages into it, it's gauranteed to block
+ //
+ log.debug("// feed more messages in order to threshold block - as Bitset requires linearly as much space as we are feeding messages into it, it's guaranteed to block");
int extraRecordsToBlockWithThresholdBlocks = numberOfRecords / 2;
{
assertThat(wm.getPm().isAllowedMoreRecords(topicPartition)).isTrue(); // should initially be not blocked
@@ -164,12 +189,12 @@ void backPressureShouldPreventTooManyMessagesBeingQueuedForProcessing() throws O
ktu.send(consumerSpy, ktu.generateRecords(extraRecordsToBlockWithThresholdBlocks));
awaitForOneLoopCycle();
- // assert partition now blocked from threshold
- waitAtMost(ofSeconds(30))
+ log.debug("// assert partition now blocked from threshold");
+ waitAtMost(ofSeconds(10))
.untilAsserted(
- () -> assertThat(wm.getPm().isBlocked(topicPartition))
- .as("Partition SHOULD be blocked due to back pressure")
- .isTrue()); // blocked
+ () -> assertWithMessage("Partition SHOULD be blocked due to back pressure")
+ .that(partitionState)
+ .isBlocked()); // blocked
Long partitionOffsetHighWaterMarks = wm.getPm().getHighestSeenOffset(topicPartition);
assertThat(partitionOffsetHighWaterMarks)
@@ -178,7 +203,7 @@ void backPressureShouldPreventTooManyMessagesBeingQueuedForProcessing() throws O
parallelConsumer.requestCommitAsap();
awaitForOneLoopCycle();
- // assert blocked, but can still write payload
+ log.debug("// assert blocked, but can still write payload");
// assert the committed offset metadata contains a payload
waitAtMost(defaultTimeout).untilAsserted(() ->
{
@@ -191,30 +216,42 @@ 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);
- Truth8.assertThat(incompletes.getHighestSeenOffset()).hasValue(numberOfRecords + extraRecordsToBlockWithThresholdBlocks - 1);
+ int expectedHighestSeen = numberOfRecords + extraRecordsToBlockWithThresholdBlocks - 1;
+ Truth8.assertThat(incompletes.getHighestSeenOffset()).hasValue(expectedHighestSeen);
}
);
}
- // test max payload exceeded, payload dropped
- int processedBeforePartitionBlock = userFuncFinishedCount.get();
- int extraMessages = numberOfRecords + extraRecordsToBlockWithThresholdBlocks / 2;
+ // recreates the situation where the payload size is too large and must be dropped
+ log.debug("// test max payload exceeded, payload dropped");
{
- // force system to allow more records (i.e. the actual system attempts to never allow the payload to grow this big)
- PartitionMonitor.setUSED_PAYLOAD_THRESHOLD_MULTIPLIER(2);
+ log.debug("// force system to allow more records (i.e. the actual system attempts to never allow the payload to grow this big)");
+ PartitionStateManager.setUSED_PAYLOAD_THRESHOLD_MULTIPLIER(30);
+ parallelConsumer.requestCommitAsap();
+ awaitForOneLoopCycle();
+
//
- msgLockThree.countDown(); // unlock to make state dirty to get a commit
+ log.debug("// unlock to make state dirty to get a commit");
+
+ msgLockThree.countDown();
+ int processedBeforePartitionBlock = userFuncFinishedCount.get();
+ int extraMessages = numberOfRecords + extraRecordsToBlockWithThresholdBlocks / 2;
+ log.debug("// send {} more messages", extraMessages);
ktu.send(consumerSpy, ktu.generateRecords(extraMessages));
awaitForOneLoopCycle();
parallelConsumer.requestCommitAsap();
- //
+ log.debug("// wait for the new message to be processed");
await().atMost(defaultTimeout).untilAsserted(() ->
- assertThat(userFuncFinishedCount.get()).isEqualTo(processedBeforePartitionBlock + extraMessages + 1) // some new message processed
+ {
+ int expectedUserFunctionFinishedCount = processedBeforePartitionBlock + extraMessages + 1;
+ assertThat(userFuncFinishedCount.get()).isEqualTo(expectedUserFunctionFinishedCount);
+ }
);
- // assert payload missing from commit now
+
+ log.debug("// assert payload missing from commit now");
await().untilAsserted(() -> {
OffsetAndMetadata partitionCommit = getLastCommit();
assertThat(partitionCommit.offset()).isZero();
@@ -222,7 +259,7 @@ void backPressureShouldPreventTooManyMessagesBeingQueuedForProcessing() throws O
});
}
- // test failed messages can retry
+ log.debug("// test failed messages can retry");
{
Duration aggressiveDelay = ofMillis(100);
WorkContainer.setDefaultRetryDelay(aggressiveDelay); // more aggressive retry
@@ -259,10 +296,6 @@ void backPressureShouldPreventTooManyMessagesBeingQueuedForProcessing() throws O
await().untilAsserted(() -> assertThat(wm.getPm().isAllowedMoreRecords(topicPartition)).isTrue());
}
} finally {
- // make sure to unlock threads - speeds up failed tests, instead of waiting for latch or close timeouts
- msgLock.countDown();
- msgLockTwo.countDown();
-
// todo restore static defaults - lazy way to override settings at runtime but causes bugs by allowing them to be statically changeable
OffsetMapCodecManager.DefaultMaxMetadataSize = realMax; // todo wow this is smelly, but convenient
OffsetMapCodecManager.forcedCodec = Optional.empty();
diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingBackPressureUnitTest.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingBackPressureUnitTest.java
new file mode 100644
index 000000000..7d420d5a2
--- /dev/null
+++ b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingBackPressureUnitTest.java
@@ -0,0 +1,221 @@
+package io.confluent.parallelconsumer.offsets;
+
+/*-
+ * Copyright (C) 2020-2022 Confluent, Inc.
+ */
+
+import com.google.common.truth.Truth;
+import io.confluent.parallelconsumer.ParallelEoSStreamProcessorTestBase;
+import io.confluent.parallelconsumer.internal.EpochAndRecordsMap;
+import io.confluent.parallelconsumer.state.PartitionState;
+import io.confluent.parallelconsumer.state.PartitionStateManager;
+import io.confluent.parallelconsumer.state.WorkContainer;
+import io.confluent.parallelconsumer.state.WorkManager;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import one.util.streamex.LongStreamEx;
+import one.util.streamex.StreamEx;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.ResourceAccessMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+import pl.tlinkowski.unij.api.UniLists;
+import pl.tlinkowski.unij.api.UniMaps;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import static io.confluent.parallelconsumer.ManagedTruth.assertTruth;
+import static io.confluent.parallelconsumer.ManagedTruth.assertWithMessage;
+import static io.confluent.parallelconsumer.state.PartitionStateManager.USED_PAYLOAD_THRESHOLD_MULTIPLIER_DEFAULT;
+import static java.time.Duration.ofMillis;
+
+/**
+ * UnitTest version of {@link OffsetEncodingBackPressureTest}.
+ *
+ * @see OffsetEncodingBackPressureTest
+ */
+@Slf4j
+class OffsetEncodingBackPressureUnitTest extends ParallelEoSStreamProcessorTestBase {
+
+ @AfterAll
+ static void cleanup() {
+ PartitionStateManager.setUSED_PAYLOAD_THRESHOLD_MULTIPLIER(USED_PAYLOAD_THRESHOLD_MULTIPLIER_DEFAULT);
+ }
+
+ @SneakyThrows
+ @Test
+ @ResourceLock(value = OffsetMapCodecManager.METADATA_DATA_SIZE_RESOURCE_LOCK, mode = ResourceAccessMode.READ_WRITE)
+ void backPressureShouldPreventTooManyMessagesBeingQueuedForProcessing() throws OffsetDecodingError {
+ final int numberOfRecords = 1_00;
+
+ // todo - very smelly - store for restoring
+ var realMax = OffsetMapCodecManager.DefaultMaxMetadataSize;
+
+ // todo don't use static public accessors to change things - makes parallel testing harder and is smelly
+ OffsetMapCodecManager.DefaultMaxMetadataSize = 40; // reduce available to make testing easier
+ OffsetMapCodecManager.forcedCodec = Optional.of(OffsetEncoding.BitSetV2); // force one that takes a predictable large amount of space
+
+ //
+ var wm = parallelConsumer.getWm();
+ var pm = wm.getPm();
+ PartitionState partitionState = pm.getPartitionState(topicPartition);
+
+ sendRecordsToWM(numberOfRecords, wm);
+
+ final int numberOfBlockedMessages = 2;
+ var samplingOfShouldBeCompleteOffsets = UniLists.of(1L, 50L, 99L, (long) numberOfRecords - numberOfBlockedMessages);
+ var blockedOffsets = UniLists.of(0L, 2L);
+
+ var completes = LongStreamEx.of(numberOfRecords).filter(x -> !blockedOffsets.contains(x)).boxed().toList();
+
+ List> workIfAvailable = wm.getWorkIfAvailable();
+ assertTruth(workIfAvailable).hasSize(numberOfRecords);
+
+ List> toSucceed = workIfAvailable.stream().filter(x -> !blockedOffsets.contains(x.offset())).collect(Collectors.toList());
+ toSucceed.forEach(wm::onSuccessResult);
+
+ try {
+
+ // # assert commit ok - nothing blocked
+ {
+ Optional commitDataIfDirty = partitionState.getCommitDataIfDirty();
+ assertTruth(partitionState).isAllowedMoreRecords();
+
+ int expectedHighestSeenOffset = numberOfRecords - 1;
+ // check("getOffsetHighestSucceeded()").that(actual.getOffsetHighestSucceeded()).isEqualTo(expected);
+ assertTruth(partitionState).getOffsetHighestSeen().isEqualTo(expectedHighestSeenOffset);
+ assertTruth(partitionState).getCommitDataIfDirty().hasOffsetEqualTo(0);
+ }
+
+
+ log.debug("// feed more messages in order to threshold block - as Bitset requires linearly as much space as we are feeding messages into it, it's guaranteed to block");
+ int extraRecordsToBlockWithThresholdBlocks = numberOfRecords / 2;
+ {
+ sendRecordsToWM(extraRecordsToBlockWithThresholdBlocks, wm);
+ succeedExcept(wm, blockedOffsets);
+
+ // triggers recompute of blockage
+ Optional commitDataIfDirty = partitionState.getCommitDataIfDirty();
+
+ log.debug("// assert partition now blocked from threshold");
+ assertTruth(partitionState).isNotAllowedMoreRecords();
+
+ log.debug("// assert blocked, but can still write payload");
+ assertTruth(partitionState).getCommitDataIfDirty().hasOffsetEqualTo(0L);
+
+ // "The only incomplete record now is offset zero, which we are blocked on"
+ assertTruth(partitionState).getOffsetHighestSeen().isEqualTo(numberOfRecords + extraRecordsToBlockWithThresholdBlocks - 1);
+ assertTruth(partitionState).getCommitDataIfDirty().getMetadata().isNotEmpty();
+ assertTruth(partitionState)
+ .getAllIncompleteOffsets()
+ .containsNoneIn(samplingOfShouldBeCompleteOffsets);
+ assertWithMessage("The only incomplete record now is offset zero, which we are blocked on")
+ .that(partitionState).getAllIncompleteOffsets().containsExactlyElementsIn(blockedOffsets);
+ }
+
+
+ // recreates the situation where the payload size is too large and must be dropped
+ log.debug("// test max payload exceeded, payload dropped");
+ int processedBeforePartitionBlock = extraRecordsToBlockWithThresholdBlocks + numberOfRecords - blockedOffsets.size();
+ int extraMessages = numberOfRecords + extraRecordsToBlockWithThresholdBlocks / 2;
+ log.debug("// messages already sent {}, sending {} more", processedBeforePartitionBlock, extraMessages);
+ {
+ log.debug("// force system to allow more records (i.e. the actual system attempts to never allow the payload to grow this big)");
+ PartitionStateManager.setUSED_PAYLOAD_THRESHOLD_MULTIPLIER(2);
+
+ //
+ // unlock 2L as well
+ unblock(wm, workIfAvailable, 2L);
+ log.debug("// unlock to make state dirty to get a commit");
+ Optional commitDataIfDirty = partitionState.getCommitDataIfDirty();
+
+ //
+ log.debug("// send {} more messages", extraMessages);
+ sendRecordsToWM(extraMessages, wm);
+ succeedExcept(wm, UniLists.of(0L));
+
+ log.debug("// assert payload missing from commit now");
+
+ assertTruth(partitionState).getCommitDataIfDirty().hasOffsetEqualTo(0);
+ assertTruth(partitionState).getCommitDataIfDirty().getMetadata().isEmpty();
+ }
+
+ log.debug("// test failed messages can retry");
+ {
+ {
+ // check it's not returned
+ List workIfAvailable1 = StreamEx.of(wm.getWorkIfAvailable()).map(WorkContainer::offset).toList();
+ assertTruth(workIfAvailable1).doesNotContain(0L);
+ }
+
+ // more aggressive retry
+ Duration aggressiveDelay = ofMillis(100);
+ WorkContainer.setDefaultRetryDelay(aggressiveDelay);
+
+ // release message that was blocking partition progression
+
+ wm.onFailureResult(findWC(workIfAvailable, 0L));
+
+ {
+ List workIfAvailable1 = StreamEx.of(wm.getWorkIfAvailable()).map(WorkContainer::offset).toList();
+ assertTruth(workIfAvailable1).contains(0L);
+ }
+
+
+ unblock(wm, workIfAvailable, 0L);
+ }
+
+ // assert partition is now not blocked
+ {
+
+ Optional commitDataIfDirty = partitionState.getCommitDataIfDirty();
+ assertTruth(partitionState).isAllowedMoreRecords();
+
+ }
+
+ // assert all committed, nothing blocked- next expected offset is now 1+ the offset of the final message we sent
+ {
+
+ assertTruth(partitionState).getCommitDataIfDirty().getOffset().isEqualTo(processedBeforePartitionBlock + extraMessages + numberOfBlockedMessages);
+
+ assertTruth(partitionState).isAllowedMoreRecords();
+
+ }
+ } finally {
+ // todo restore static defaults - lazy way to override settings at runtime but causes bugs by allowing them to be statically changeable
+ OffsetMapCodecManager.DefaultMaxMetadataSize = realMax; // todo wow this is smelly, but convenient
+ OffsetMapCodecManager.forcedCodec = Optional.empty();
+ }
+ }
+
+ private void succeedExcept(WorkManager wm, List incomplete) {
+ var workIfAvailable = wm.getWorkIfAvailable();
+ var toSucceed = workIfAvailable.stream()
+ .filter(x -> !incomplete.contains(x.offset()))
+ .collect(Collectors.toList());
+ toSucceed.forEach(wm::onSuccessResult);
+ }
+
+ private void unblock(WorkManager wm, List> from, long offsetToUnblock) {
+ var unblock = findWC(from, offsetToUnblock);
+ wm.onSuccessResult(unblock);
+ }
+
+ private WorkContainer findWC(List> from, long offsetToUnblock) {
+ return from.stream().filter(x -> x.offset() == offsetToUnblock).findFirst().get();
+ }
+
+ private void sendRecordsToWM(int numberOfRecords, WorkManager wm) {
+ log.debug("~Sending {} more records", numberOfRecords);
+ List> records = ktu.generateRecords(numberOfRecords);
+ wm.registerWork(new EpochAndRecordsMap<>(new ConsumerRecords<>(UniMaps.of(topicPartition, records)), wm.getPm()));
+ Truth.assertThat(wm.getNumberOfWorkQueuedInShardsAwaitingSelection()).isEqualTo(numberOfRecords);
+ }
+
+}
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 0507a78ec..cd2a0d866 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
@@ -8,6 +8,7 @@
import io.confluent.csid.utils.KafkaTestUtils;
import io.confluent.parallelconsumer.ParallelConsumerOptions;
import io.confluent.parallelconsumer.ParallelEoSStreamProcessorTestBase;
+import io.confluent.parallelconsumer.internal.EpochAndRecordsMap;
import io.confluent.parallelconsumer.state.WorkContainer;
import io.confluent.parallelconsumer.state.WorkManager;
import lombok.SneakyThrows;
@@ -27,6 +28,7 @@
import java.util.*;
import java.util.stream.Collectors;
+import static io.confluent.parallelconsumer.ManagedTruth.assertTruth;
import static io.confluent.parallelconsumer.offsets.OffsetEncoding.*;
import static org.assertj.core.api.Assertions.assertThat;
import static org.hamcrest.Matchers.in;
@@ -170,14 +172,14 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset
ParallelConsumerOptions options = parallelConsumer.getWm().getOptions();
HashMap>> recordsMap = new HashMap<>();
TopicPartition tp = new TopicPartition(INPUT_TOPIC, 0);
- recordsMap.put(tp, records);
+ recordsMap.put(tp, new ArrayList<>(records));
ConsumerRecords testRecords = new ConsumerRecords<>(recordsMap);
// write offsets
{
WorkManager wmm = new WorkManager<>(options, consumerSpy);
wmm.onPartitionsAssigned(UniSets.of(new TopicPartition(INPUT_TOPIC, 0)));
- wmm.registerWork(testRecords);
+ wmm.registerWork(new EpochAndRecordsMap<>(testRecords, wmm.getPm()));
List> work = wmm.getWorkIfAvailable();
assertThat(work).hasSameSizeAs(records);
@@ -211,7 +213,7 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset
var committed = consumerSpy.committed(UniSets.of(tp)).get(tp);
assertThat(committed.offset()).isEqualTo(1L);
- if (!encodingsThatFail.contains(encoding)) {
+ if (assumeWorkingCodec(encoding, encodingsThatFail)) {
assertThat(committed.metadata()).isNotBlank();
}
}
@@ -222,12 +224,12 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset
{
var newWm = new WorkManager<>(options, consumerSpy);
newWm.onPartitionsAssigned(UniSets.of(tp));
- newWm.registerWork(testRecords);
+ newWm.registerWork(new EpochAndRecordsMap(testRecords, newWm.getPm()));
var pm = newWm.getPm();
var partitionState = pm.getPartitionState(tp);
- if (!encodingsThatFail.contains(encoding)) {
+ if (assumeWorkingCodec(encoding, encodingsThatFail)) {
long offsetHighestSequentialSucceeded = partitionState.getOffsetHighestSequentialSucceeded();
assertThat(offsetHighestSequentialSucceeded).isEqualTo(0);
@@ -245,11 +247,9 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset
var anIncompleteRecord = records.get(3);
Truth.assertThat(pm.isRecordPreviouslyCompleted(anIncompleteRecord)).isFalse();
- // force ingestion early, and check results
+ // check state
{
- int ingested = newWm.tryToEnsureQuantityOfWorkQueuedAvailable(Integer.MAX_VALUE);
-
- if (!encodingsThatFail.contains(encoding)) {
+ if (assumeWorkingCodec(encoding, encodingsThatFail)) {
long offsetHighestSequentialSucceeded = partitionState.getOffsetHighestSequentialSucceeded();
assertThat(offsetHighestSequentialSucceeded).isEqualTo(0);
@@ -262,7 +262,6 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset
var incompletes = partitionState.getIncompleteOffsetsBelowHighestSucceeded();
Truth.assertThat(incompletes).containsExactlyElementsIn(expected);
- assertThat(ingested).isEqualTo(testRecords.count() - 4); // 4 were succeeded
Truth.assertThat(pm.isRecordPreviouslyCompleted(anIncompleteRecord)).isFalse();
}
}
@@ -270,7 +269,7 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset
var workRetrieved = newWm.getWorkIfAvailable();
var workRetrievedOffsets = workRetrieved.stream().map(WorkContainer::offset).collect(Collectors.toList());
- Truth.assertThat(workRetrieved).isNotEmpty();
+ assertTruth(workRetrieved).isNotEmpty();
switch (encoding) {
case BitSet, BitSetCompressed, // BitSetV1 both get a short overflow due to the length being too long
@@ -281,9 +280,10 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset
assertThat(workRetrievedOffsets).doesNotContainSequence(expected);
}
default -> {
- assertThat(workRetrievedOffsets)
- .as("Contains only incomplete records")
- .containsExactlyElementsOf(expected);
+ Truth.assertWithMessage("Contains only incomplete records")
+ .that(workRetrievedOffsets)
+ .containsExactlyElementsIn(expected)
+ .inOrder();
}
}
}
@@ -291,6 +291,13 @@ void ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential(Offset
OffsetSimultaneousEncoder.compressionForced = false;
}
+ /**
+ * A {@link OffsetEncoding} that works in this test scenario
+ */
+ private boolean assumeWorkingCodec(OffsetEncoding encoding, List encodingsThatFail) {
+ return !encodingsThatFail.contains(encoding);
+ }
+
/**
* This version of non sequential test just test the encoder directly, and is only half the story, as at the
* encoding stage they don't know which offsets have never been seen, and assume simply working with continuous
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 e284bcd0c..afb6ff16c 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
@@ -9,7 +9,9 @@
import io.confluent.csid.utils.LongPollingMockConsumer;
import io.confluent.csid.utils.TimeUtils;
import io.confluent.parallelconsumer.FakeRuntimeError;
+import io.confluent.parallelconsumer.ManagedTruth;
import io.confluent.parallelconsumer.ParallelConsumerOptions;
+import io.confluent.parallelconsumer.internal.EpochAndRecordsMap;
import io.confluent.parallelconsumer.truth.CommitHistorySubject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -29,14 +31,16 @@
import org.junit.jupiter.params.provider.ValueSource;
import org.threeten.extra.MutableClock;
import pl.tlinkowski.unij.api.UniLists;
+import pl.tlinkowski.unij.api.UniMaps;
import java.time.Duration;
import java.util.*;
+import java.util.stream.Collectors;
+import static com.google.common.truth.Truth.assertWithMessage;
import static io.confluent.csid.utils.Range.range;
import static io.confluent.parallelconsumer.ParallelConsumerOptions.ProcessingOrder.*;
import static java.time.Duration.ofSeconds;
-import static java.util.Comparator.comparingLong;
import static org.assertj.core.api.Assertions.assertThat;
import static pl.tlinkowski.unij.api.UniLists.of;
@@ -70,32 +74,37 @@ private void setupWorkManager(ParallelConsumerOptions build) {
log.debug("Heard some successful work: {}", work);
successfulWork.add(work);
});
- int partition = 0;
- assignPartition(partition);
+
}
private void assignPartition(final int partition) {
- wm.onPartitionsAssigned(UniLists.of(getTopicPartition(partition)));
+ wm.onPartitionsAssigned(UniLists.of(topicPartitionOf(partition)));
}
@NotNull
- private TopicPartition getTopicPartition(int partition) {
+ private TopicPartition topicPartitionOf(int partition) {
return new TopicPartition(INPUT_TOPIC, partition);
}
+ private void registerSomeWork() {
+ registerSomeWork(0);
+ }
+
/**
* Adds 3 units of work
*/
- private void registerSomeWork() {
+ private void registerSomeWork(int partition) {
+ assignPartition(partition);
+
String key = "key-0";
- int partition = 0;
+
var rec0 = makeRec("0", key, partition);
var rec1 = makeRec("1", key, partition);
var rec2 = makeRec("2", key, partition);
Map>> m = new HashMap<>();
- m.put(getTopicPartition(partition), of(rec0, rec1, rec2));
+ m.put(topicPartitionOf(partition), of(rec0, rec1, rec2));
var recs = new ConsumerRecords<>(m);
- wm.registerWork(recs);
+ wm.registerWork(new EpochAndRecordsMap(recs, wm.getPm()));
}
private ConsumerRecord makeRec(String value, String key, int partition) {
@@ -104,21 +113,41 @@ private ConsumerRecord makeRec(String value, String key, int par
return stringStringConsumerRecord;
}
- @Test
- void testRemovedUnordered() {
- setupWorkManager(ParallelConsumerOptions.builder().ordering(UNORDERED).build());
+ @ParameterizedTest
+ @EnumSource
+ void basic(ParallelConsumerOptions.ProcessingOrder order) {
+ setupWorkManager(ParallelConsumerOptions.builder()
+ .ordering(order)
+ .build());
registerSomeWork();
- int max = 1;
- var gottenWork = wm.getWorkIfAvailable(max);
- assertThat(gottenWork).hasSize(1);
- assertOffsets(gottenWork, of(0));
+ //
+ var gottenWork = wm.getWorkIfAvailable();
+
+ if (order == UNORDERED) {
+ assertThat(gottenWork).hasSize(3);
+ assertOffsets(gottenWork, of(0, 1, 2));
+ } else {
+ assertThat(gottenWork).hasSize(1);
+ assertOffsets(gottenWork, of(0));
+ }
+ //
wm.onSuccessResult(gottenWork.get(0));
- gottenWork = wm.getWorkIfAvailable(max);
- assertThat(gottenWork).hasSize(1);
- assertOffsets(gottenWork, of(1));
+ //
+ gottenWork = wm.getWorkIfAvailable();
+
+ if (order == UNORDERED) {
+ assertThat(gottenWork).isEmpty();
+ } else {
+ assertThat(gottenWork).hasSize(1);
+ assertOffsets(gottenWork, of(1));
+ }
+
+ //
+ gottenWork = wm.getWorkIfAvailable();
+ assertThat(gottenWork).isEmpty();
}
@Test
@@ -336,11 +365,11 @@ 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(getTopicPartition(partition), of(rec2, rec3, rec));
+ m.put(topicPartitionOf(partition), of(rec2, rec3, rec));
var recs = new ConsumerRecords<>(m);
//
- wm.registerWork(recs);
+ registerWork(recs);
int max = 10;
@@ -363,6 +392,11 @@ void insertWrongOrderPreservesOffsetOrdering() {
assertOffsets(works, of(1, 6));
}
+ private void registerWork(ConsumerRecords recs) {
+ wm.registerWork(new EpochAndRecordsMap<>(recs, wm.getPm()));
+ }
+
+
private void fail(WorkContainer wc) {
wc.onUserFunctionFailure(null);
wm.onFailureResult(wc);
@@ -414,22 +448,6 @@ public int size() {
}
}
- @Test
- @Disabled
- public void multipleFailures() {
- }
-
-
- @Test
- @Disabled
- public void delayedOrdered() {
- }
-
- @Test
- @Disabled
- public void delayedUnordered() {
- }
-
@Test
void orderedByPartitionsParallel() {
ParallelConsumerOptions build = ParallelConsumerOptions.builder()
@@ -445,11 +463,11 @@ 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(getTopicPartition(partition), of(rec2, rec3, rec));
+ m.put(topicPartitionOf(partition), of(rec2, rec3, rec));
var recs = new ConsumerRecords<>(m);
//
- wm.registerWork(recs);
+ registerWork(recs);
//
var works = wm.getWorkIfAvailable();
@@ -491,11 +509,11 @@ 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(getTopicPartition(partition), of(rec2, rec3, rec0, rec4, rec5, rec6));
+ m.put(topicPartitionOf(partition), of(rec2, rec3, rec0, rec4, rec5, rec6));
var recs = new ConsumerRecords<>(m);
//
- wm.registerWork(recs);
+ registerWork(recs);
//
var works = wm.getWorkIfAvailable();
@@ -520,18 +538,14 @@ void orderedByKeyParallel() {
assertOffsets(works, of());
}
- @Test
- @Disabled
- public void unorderedPartitionsGreedy() {
- }
-
- // @Test
@ParameterizedTest
@ValueSource(ints = {1, 2, 5, 10, 20, 30, 50, 1000})
void highVolumeKeyOrder(int quantity) {
int uniqueKeys = 100;
- var build = ParallelConsumerOptions.builder().ordering(KEY).build();
+ var build = ParallelConsumerOptions.builder()
+ .ordering(KEY)
+ .build();
setupWorkManager(build);
KafkaTestUtils ktu = new KafkaTestUtils(INPUT_TOPIC, null, new LongPollingMockConsumer<>(OffsetResetStrategy.EARLIEST));
@@ -540,20 +554,24 @@ void highVolumeKeyOrder(int quantity) {
var records = ktu.generateRecords(keys, quantity);
var flattened = ktu.flatten(records.values());
- flattened.sort(comparingLong(ConsumerRecord::offset));
- Map>> m = new HashMap<>();
- m.put(getTopicPartition(0), flattened);
- var recs = new ConsumerRecords<>(m);
+ int partition = 0;
+ var recs = new ConsumerRecords<>(UniMaps.of(topicPartitionOf(partition), flattened));
+
+ assignPartition(partition);
+
+ //
+ registerWork(recs);
//
- wm.registerWork(recs);
+ long awaiting = wm.getSm().getNumberOfWorkQueuedInShardsAwaitingSelection();
+ assertThat(awaiting).isEqualTo(quantity);
//
List> work = wm.getWorkIfAvailable();
//
- assertThat(work).hasSameSizeAs(records.keySet());
+ ManagedTruth.assertTruth(work).hasSameSizeAs(records);
}
@Test
@@ -603,7 +621,7 @@ void workQueuesEmptyWhenAllWorkComplete() {
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));
+ PartitionState state = wm.getPm().getPartitionState(topicPartitionOf(0));
Truth.assertThat(state.getAllIncompleteOffsets()).isEmpty();
}
@@ -626,14 +644,14 @@ void resumesFromNextShard(ParallelConsumerOptions.ProcessingOrder order) {
assignPartition(2);
Map>> m = new HashMap<>();
var rec = new ConsumerRecord<>(INPUT_TOPIC, 1, 11, "11", "value");
- m.put(getTopicPartition(1), of(rec));
+ m.put(topicPartitionOf(1), of(rec));
var rec2 = new ConsumerRecord<>(INPUT_TOPIC, 2, 21, "21", "value");
- m.put(getTopicPartition(2), of(rec2));
+ m.put(topicPartitionOf(2), of(rec2));
var recs = new ConsumerRecords<>(m);
- wm.registerWork(recs);
+ registerWork(recs);
- // force ingestion of records - see refactor: Queue unification #219
- wm.tryToEnsureQuantityOfWorkQueuedAvailable(100);
+// // force ingestion of records - see refactor: Queue unification #219
+// wm.tryToEnsureQuantityOfWorkQueuedAvailable(100);
var workContainersOne = wm.getWorkIfAvailable(1);
var workContainersTwo = wm.getWorkIfAvailable(1);
@@ -658,4 +676,57 @@ void resumesFromNextShard(ParallelConsumerOptions.ProcessingOrder order) {
}
}
+
+ /**
+ * Checks that when using shards are not starved when there's enough work queued to satisfy poll request from the
+ * initial request (without needing to iterate to other shards)
+ *
+ * @see #236 Under some conditions, a
+ * shard (by partition or key), can get starved for attention
+ */
+ @Test
+ void starvation() {
+ setupWorkManager(ParallelConsumerOptions.builder()
+ .ordering(PARTITION)
+ .build());
+
+ registerSomeWork(0);
+ registerSomeWork(1);
+ registerSomeWork(2);
+
+ var allWork = new ArrayList>();
+
+ {
+ var work = wm.getWorkIfAvailable(2);
+ allWork.addAll(work);
+
+ assertWithMessage("Should be able to get 2 records of work, one from each partition shard")
+ .that(work).hasSize(2);
+
+ //
+ var tpOne = work.get(0).getTopicPartition();
+ var tpTwo = work.get(1).getTopicPartition();
+ assertWithMessage("The partitions should be different")
+ .that(tpOne).isNotEqualTo(tpTwo);
+
+ }
+
+ {
+ var work = wm.getWorkIfAvailable(2);
+ assertWithMessage("Should be able to get only 1 more, from the third shard")
+ .that(work).hasSize(1);
+ allWork.addAll(work);
+
+ //
+ var tpOne = work.get(0).getTopicPartition();
+ }
+
+ assertWithMessage("TPs all unique")
+ .that(allWork.stream()
+ .map(WorkContainer::getTopicPartition)
+ .collect(Collectors.toList()))
+ .containsNoDuplicates();
+
+ }
+
}
diff --git a/parallel-consumer-core/src/test/resources/logback-test.xml b/parallel-consumer-core/src/test/resources/logback-test.xml
index 906410547..e7d4e3b4f 100644
--- a/parallel-consumer-core/src/test/resources/logback-test.xml
+++ b/parallel-consumer-core/src/test/resources/logback-test.xml
@@ -28,17 +28,22 @@
-
+
+
+
+
+
+
-
+
-
+
@@ -47,10 +52,10 @@
-
-
+
+
-
+
diff --git a/pom.xml b/pom.xml
index 0d030e132..2b9da56f3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -521,6 +521,7 @@
false
+
true