This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new 8ddcc090 [FLINK-33484][Connectors/Kafka] Correct offsets committed to
Kafka
8ddcc090 is described below
commit 8ddcc0901850bb69baaf4bf6801632947985cb7c
Author: Dale Lane <[email protected]>
AuthorDate: Thu Oct 1 23:15:40 2026 +0100
[FLINK-33484][Connectors/Kafka] Correct offsets committed to Kafka
Offsets that the Kafka source commits back to Kafka can lag the
consumer's true position, because entries that the consumer reads but
never delivers as records (e.g. transaction commit / abort markers, and
records of aborted transactions) were not accounted for. The result is
that the committed offset was stopping at the last delivered record,
while the consumer had advanced past that.
This wasn't impacting Flink processing guarantees, as restarts restore
from checkpointed offsets, not offsets committed to Kafka. External
tooling that uses the Kafka committed offset for monitoring would give
the appearance of consumer lag. On low-traffic or bursty transactional
topics, this would report lag that does not exist until new events are
received. A partition holding only aborted transactions, read from the
earliest offset, never had an offset committed at all.
This is now changed to reconcile the offset to commit with the
consumer's position:
- KafkaPartitionSplitReader now tracks, per partition, the offset of
the last record handed to the reader, and the consumer position
observed at the end of each fetch()
- At checkpoint commit, reconcileOffsetsToCommit advances the offset to
commit to the consumer position when the reader is caught up to what
it delivered, so the committed offset will reflect the non-record
entries that the consumer read past. Reconciliation is skipped for
partitions that aren't assigned and for running bounded splits
- Splits that have not emitted a record still hold a starting offset
placeholder such as EARLIEST_OFFSET, so KafkaSourceReader has no
offset to commit for them. Their partitions are passed to the split
reader separately, which commits the consumer position for them
unless records have been fetched that the checkpoint may not include
- The split reader remembers the offset it last committed for each
assigned partition, and leaves a partition out of a commit rather
than move its committed offset backwards. This can otherwise happen
when a record is fetched between a checkpoint and its commit, so the
offset can no longer be advanced. The offset is recorded when the
commit is sent, so that a later commit cannot overtake one that is
still in flight
- A finished bounded split is committed at its stopping offset, rather
than after its last emitted record, as it only finishes once the
consumer position has reached the stopping offset. This only applies
once the split holds a real offset: a split that is still at a
starting offset placeholder may have been empty when it was assigned,
with the consumer already past its stopping offset, so committing the
stopping offset could move the offset backwards
- Per-partition tracking state is cleaned up when partitions are
unassigned
- KafkaSourceReader records the committed-offset metric from the
offsets that are committed (rather than those requested) so Kafka and
Flink's committedOffsets metrics agree.
Generated-by: Claude Code (Sonnet 5)
Signed-off-by: Dale Lane <[email protected]>
---
.../source/reader/KafkaPartitionSplitReader.java | 138 +++-
.../kafka/source/reader/KafkaSourceReader.java | 57 +-
.../reader/fetcher/KafkaSourceFetcherManager.java | 22 +-
.../reader/KafkaPartitionSplitReaderTest.java | 315 +++++++++
.../kafka/source/reader/KafkaSourceReaderTest.java | 729 ++++++++++++++++++++-
5 files changed, 1244 insertions(+), 17 deletions(-)
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
index b87ce1ce..d58a2f3d 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
@@ -86,6 +86,19 @@ public class KafkaPartitionSplitReader
// Tracking empty splits that has not been added to finished splits in
fetch()
private final Set<String> emptySplits = new HashSet<>();
+ // Offset of the last record that fetch() handed to the source reader, per
partition
+ final Map<TopicPartition, Long> lastFetchedOffsets = new HashMap<>();
+
+ // Consumer position observed at the end of the most recent fetch(), per
partition.
+ final Map<TopicPartition, Long> lastKnownPositions = new HashMap<>();
+
+ // Offset most recently sent to Kafka in a commit, per assigned partition.
Recorded when the
+ // commit is sent (instead of when it completes) so later commits won't
overtake commits in
+ // flight with a lower offset.
+ // If a commit fails, a later checkpoint with an offset lower than this
won't be committed, so
+ // the committed offset is only updated again once a checkpoint reaches it.
+ final Map<TopicPartition, Long> lastCommittedOffsets = new HashMap<>();
+
public KafkaPartitionSplitReader(
Properties props,
SourceReaderContext context,
@@ -173,6 +186,7 @@ public class KafkaPartitionSplitReader
long stoppingOffset = getStoppingOffset(tp);
long consumerPosition =
getConsumerPosition(consumer, tp, "retrieving consumer
position");
+ lastKnownPositions.put(tp, consumerPosition);
// Stop fetching when the consumer's position reaches the
stoppingOffset.
// Control messages may follow the last record; therefore, using
the last record's
// offset as a stopping condition could result in indefinite
blocking.
@@ -196,6 +210,8 @@ public class KafkaPartitionSplitReader
kafkaSourceReaderMetrics.maybeAddRecordsLagMetric(consumer, trackTp);
});
+ trackLastFetchedRecordOffsets(consumerRecords);
+
markEmptySplitsAsFinished(recordsBySplits);
// Unassign the partitions that has finished.
@@ -335,10 +351,35 @@ public class KafkaPartitionSplitReader
// ---------------
+ /**
+ * Commits the offsets of a completed checkpoint to Kafka.
+ *
+ * @param offsetsToCommit offsets that the checkpoint holds for its
partitions
+ * @param partitionsWithoutOffset partitions whose split had not emitted
any record when the
+ * checkpoint was taken, so the checkpoint holds no offset for them,
only a starting offset
+ * placeholder such as {@link KafkaPartitionSplit#EARLIEST_OFFSET}
+ * @param offsetCommitCallback called with offsets that were committed
+ */
public void notifyCheckpointComplete(
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
+ Set<TopicPartition> partitionsWithoutOffset,
OffsetCommitCallback offsetCommitCallback) {
- ensureConsumer().commitAsync(offsetsToCommit, offsetCommitCallback);
+ final KafkaConsumer<byte[], byte[]> consumer = ensureConsumer();
+ final Map<TopicPartition, OffsetAndMetadata> reconciled =
+ reconcileOffsetsToCommit(consumer, offsetsToCommit,
partitionsWithoutOffset);
+ if (reconciled.isEmpty()) {
+ // nothing left to commit after reconciling with the consumer
position
+ offsetCommitCallback.onComplete(reconciled, null);
+ return;
+ }
+ final Set<TopicPartition> assignment = consumer.assignment();
+ reconciled.forEach(
+ (tp, offsetAndMetadata) -> {
+ if (assignment.contains(tp)) {
+ lastCommittedOffsets.put(tp,
offsetAndMetadata.offset());
+ }
+ });
+ consumer.commitAsync(reconciled, offsetCommitCallback);
}
@VisibleForTesting
@@ -389,6 +430,95 @@ public class KafkaPartitionSplitReader
return retryOnWakeup(() -> consumer.position(tp), msg);
}
+ private void trackLastFetchedRecordOffsets(ConsumerRecords<byte[], byte[]>
consumerRecords) {
+ for (TopicPartition tp : consumerRecords.partitions()) {
+ List<ConsumerRecord<byte[], byte[]>> partitionRecords =
consumerRecords.records(tp);
+ if (!partitionRecords.isEmpty()) {
+ lastFetchedOffsets.put(
+ tp, partitionRecords.get(partitionRecords.size() -
1).offset());
+ }
+ }
+ }
+
+ /**
+ * Advances the offsets to commit over the entries that the Kafka consumer
read but never
+ * delivered, such as transaction control markers and records of aborted
transactions.
+ *
+ * <p>{@link KafkaRecordEmitter} derives the offset to commit from the
records it receives, so
+ * the offset stops at the first entry that Kafka does not deliver, for as
long as the partition
+ * is idle. The consumer's own position accounts for those entries, so it
is the offset that
+ * external tooling expects to see.
+ *
+ * <p>This relies on {@link #lastKnownPositions}, populated as a side
effect of the regular
+ * {@link #fetch()} poll loop, rather than querying the consumer for the
position again here.
+ * {@link #notifyCheckpointComplete} runs on this same split fetcher
thread, but at a point
+ * outside that poll loop, so this allows us to avoid a separate blocking
call to the consumer.
+ *
+ * <p>A partition that has never delivered a record (e.g. a partition with
only aborted
+ * transactions) never gets an offset in the checkpoint when its split
starts from a placeholder
+ * such as {@link KafkaPartitionSplit#EARLIEST_OFFSET}. Such partitions
are committed at the
+ * consumer position, as long as no record has been fetched for them. The
checkpoint may not
+ * include a fetched record yet, and committing the position would skip
over it.
+ *
+ * <p>Offsets are only progressed while no newer records have been
fetched, so later checkpoints
+ * can hold the same offset as an earlier one and yet not be advanced as
far. Offsets lower than
+ * the one last committed for a partition will be left out, so that the
committed offset never
+ * moves backwards.
+ */
+ private Map<TopicPartition, OffsetAndMetadata> reconcileOffsetsToCommit(
+ KafkaConsumer<byte[], byte[]> consumer,
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
+ Set<TopicPartition> partitionsWithoutOffset) {
+ Map<TopicPartition, OffsetAndMetadata> reconciled = new
HashMap<>(offsetsToCommit);
+ Set<TopicPartition> assignment = consumer.assignment();
+ offsetsToCommit.forEach(
+ (tp, offsetAndMetadata) -> {
+ if (!assignment.contains(tp) ||
stoppingOffsets.containsKey(tp)) {
+ return;
+ }
+ Long lastFetchedOffset = lastFetchedOffsets.get(tp);
+ if (lastFetchedOffset != null
+ && offsetAndMetadata.offset() != lastFetchedOffset
+ 1) {
+ return;
+ }
+ Long position = lastKnownPositions.get(tp);
+ if (position != null && position >
offsetAndMetadata.offset()) {
+ LOG.debug(
+ "Advancing offset to commit for {} from {} to
the consumer position {}.",
+ tp,
+ offsetAndMetadata.offset(),
+ position);
+ reconciled.put(
+ tp, new OffsetAndMetadata(position,
offsetAndMetadata.metadata()));
+ }
+ });
+ partitionsWithoutOffset.forEach(
+ tp -> {
+ if (!assignment.contains(tp)
+ || stoppingOffsets.containsKey(tp)
+ || lastFetchedOffsets.containsKey(tp)) {
+ return;
+ }
+ Long position = lastKnownPositions.get(tp);
+ if (position != null) {
+ LOG.debug(
+ "Committing the consumer position {} for {}
which has not delivered any records.",
+ position,
+ tp);
+ reconciled.put(tp, new OffsetAndMetadata(position));
+ }
+ });
+ reconciled
+ .entrySet()
+ .removeIf(
+ entry -> {
+ Long lastCommittedOffset =
lastCommittedOffsets.get(entry.getKey());
+ return lastCommittedOffset != null
+ && entry.getValue().offset() <
lastCommittedOffset;
+ });
+ return reconciled;
+ }
+
private void parseStartingOffsets(
KafkaPartitionSplit split,
List<TopicPartition> partitionsStartingFromEarliest,
@@ -535,6 +665,12 @@ public class KafkaPartitionSplitReader
Collection<TopicPartition> newAssignment = new
HashSet<>(consumer.assignment());
newAssignment.removeAll(partitionsToUnassign);
consumer.assign(newAssignment);
+ partitionsToUnassign.forEach(
+ tp -> {
+ lastFetchedOffsets.remove(tp);
+ lastKnownPositions.remove(tp);
+ lastCommittedOffsets.remove(tp);
+ });
}
private String createConsumerClientId(Properties props) {
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
index 317fc809..581d2d7b 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
@@ -40,8 +40,10 @@ import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
@@ -56,6 +58,9 @@ public class KafkaSourceReader<T>
// These maps need to be concurrent because it will be accessed by both
the main thread
// and the split fetcher thread in the callback.
private final SortedMap<Long, Map<TopicPartition, OffsetAndMetadata>>
offsetsToCommit;
+ // Partitions of splits that had not emitted a record when a checkpoint
was taken, so the
+ // checkpoint has no offset to commit for them. (The split reader resolves
them instead.)
+ private final SortedMap<Long, Set<TopicPartition>>
partitionsWithoutOffsetToCommit;
private final ConcurrentMap<TopicPartition, OffsetAndMetadata>
offsetsOfFinishedSplits;
private final KafkaSourceReaderMetrics kafkaSourceReaderMetrics;
private final boolean commitOffsetsOnCheckpoint;
@@ -71,6 +76,7 @@ public class KafkaSourceReader<T>
KafkaSourceReaderMetrics kafkaSourceReaderMetrics) {
super(kafkaSourceFetcherManager, recordEmitter, config, context);
this.offsetsToCommit = Collections.synchronizedSortedMap(new
TreeMap<>());
+ this.partitionsWithoutOffsetToCommit =
Collections.synchronizedSortedMap(new TreeMap<>());
this.offsetsOfFinishedSplits = new ConcurrentHashMap<>();
this.kafkaSourceReaderMetrics = kafkaSourceReaderMetrics;
this.commitOffsetsOnCheckpoint =
@@ -86,10 +92,24 @@ public class KafkaSourceReader<T>
protected void onSplitFinished(Map<String, KafkaPartitionSplitState>
finishedSplitIds) {
finishedSplitIds.forEach(
(ignored, splitState) -> {
- if (splitState.getCurrentOffset() >= 0) {
+ long offsetToCommit = splitState.getCurrentOffset();
+ // A split read to its stopping offset has read every
entry before it (including
+ // non-record events that were not emitted). A split still
holding a starting
+ // offset placeholder may instead have been empty when it
was assigned, with the
+ // consumer already past its stopping offset, so
committing that would move the
+ // offset backwards.
+ // A stopping offset that the split reader resolves
itself, such as
+ // LATEST_OFFSET, is negative here and is never chosen.
+ if (offsetToCommit >= 0) {
+ offsetToCommit =
+ Math.max(
+ offsetToCommit,
+ splitState
+ .getStoppingOffset()
+
.orElse(KafkaPartitionSplit.NO_STOPPING_OFFSET));
offsetsOfFinishedSplits.put(
splitState.getTopicPartition(),
- new
OffsetAndMetadata(splitState.getCurrentOffset()));
+ new OffsetAndMetadata(offsetToCommit));
}
});
}
@@ -114,6 +134,10 @@ public class KafkaSourceReader<T>
offsetsMap.put(
split.getTopicPartition(),
new OffsetAndMetadata(split.getStartingOffset()));
+ } else {
+ partitionsWithoutOffsetToCommit
+ .computeIfAbsent(checkpointId, id -> new
HashSet<>())
+ .add(split.getTopicPartition());
}
}
// Put offsets of all the finished splits.
@@ -136,7 +160,9 @@ public class KafkaSourceReader<T>
return;
}
- if (committedPartitions.isEmpty()) {
+ final Set<TopicPartition> partitionsWithoutOffset =
+ partitionsWithoutOffsetToCommit.getOrDefault(checkpointId,
Collections.emptySet());
+ if (committedPartitions.isEmpty() &&
partitionsWithoutOffset.isEmpty()) {
LOG.debug("There are no offsets to commit for checkpoint {}.",
checkpointId);
removeAllOffsetsToCommitUpToCheckpoint(checkpointId);
return;
@@ -145,7 +171,8 @@ public class KafkaSourceReader<T>
((KafkaSourceFetcherManager) splitFetcherManager)
.commitOffsets(
committedPartitions,
- (ignored, e) -> {
+ partitionsWithoutOffset,
+ (committedOffsets, e) -> {
// The offset commit here is needed by the
external monitoring. It won't
// break Flink job's correctness if we fail to
commit the offset here.
if (e != null) {
@@ -158,13 +185,19 @@ public class KafkaSourceReader<T>
LOG.debug(
"Successfully committed offsets for
checkpoint {}",
checkpointId);
-
kafkaSourceReaderMetrics.recordSucceededCommit();
- // If the finished topic partition has been
committed, we remove it
- // from the offsets of the finished splits map.
- committedPartitions.forEach(
+ // nothing is sent to Kafka when none of the
partitions without
+ // an offset could be resolved
+ if (!committedOffsets.isEmpty()) {
+
kafkaSourceReaderMetrics.recordSucceededCommit();
+ }
+ // offsets committed to Kafka can differ from
what was requested,
+ // as the split reader reconciles them with
the consumer position
+ committedOffsets.forEach(
(tp, offset) ->
kafkaSourceReaderMetrics.recordCommittedOffset(
tp, offset.offset()));
+ // If the finished topic partition has been
committed, remove it
+ // from the offsets of the finished splits
offsetsOfFinishedSplits
.entrySet()
.removeIf(
@@ -180,6 +213,10 @@ public class KafkaSourceReader<T>
while (!offsetsToCommit.isEmpty() && offsetsToCommit.firstKey() <=
checkpointId) {
offsetsToCommit.remove(offsetsToCommit.firstKey());
}
+ while (!partitionsWithoutOffsetToCommit.isEmpty()
+ && partitionsWithoutOffsetToCommit.firstKey() <= checkpointId)
{
+
partitionsWithoutOffsetToCommit.remove(partitionsWithoutOffsetToCommit.firstKey());
+ }
}
@Override
@@ -199,6 +236,10 @@ public class KafkaSourceReader<T>
return offsetsToCommit;
}
+ SortedMap<Long, Set<TopicPartition>> getPartitionsWithoutOffsetToCommit() {
+ return partitionsWithoutOffsetToCommit;
+ }
+
@VisibleForTesting
int getNumAliveFetchers() {
return splitFetcherManager.getNumAliveFetchers();
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/fetcher/KafkaSourceFetcherManager.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/fetcher/KafkaSourceFetcherManager.java
index 115a0e8b..d7751751 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/fetcher/KafkaSourceFetcherManager.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/fetcher/KafkaSourceFetcherManager.java
@@ -40,6 +40,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Collection;
import java.util.Map;
+import java.util.Set;
import java.util.function.Consumer;
import java.util.function.Supplier;
@@ -73,19 +74,26 @@ public class KafkaSourceFetcherManager
}
public void commitOffsets(
- Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
OffsetCommitCallback callback) {
- LOG.debug("Committing offsets {}", offsetsToCommit);
- if (offsetsToCommit.isEmpty()) {
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
+ Set<TopicPartition> partitionsWithoutOffset,
+ OffsetCommitCallback callback) {
+ LOG.debug(
+ "Committing offsets {} and positions of {}",
+ offsetsToCommit,
+ partitionsWithoutOffset);
+ if (offsetsToCommit.isEmpty() && partitionsWithoutOffset.isEmpty()) {
return;
}
SplitFetcher<ConsumerRecord<byte[], byte[]>, KafkaPartitionSplit>
splitFetcher =
fetchers.get(0);
if (splitFetcher != null) {
// The fetcher thread is still running. This should be the
majority of the cases.
- enqueueOffsetsCommitTask(splitFetcher, offsetsToCommit, callback);
+ enqueueOffsetsCommitTask(
+ splitFetcher, offsetsToCommit, partitionsWithoutOffset,
callback);
} else {
splitFetcher = createSplitFetcher();
- enqueueOffsetsCommitTask(splitFetcher, offsetsToCommit, callback);
+ enqueueOffsetsCommitTask(
+ splitFetcher, offsetsToCommit, partitionsWithoutOffset,
callback);
startFetcher(splitFetcher);
}
}
@@ -93,6 +101,7 @@ public class KafkaSourceFetcherManager
private void enqueueOffsetsCommitTask(
SplitFetcher<ConsumerRecord<byte[], byte[]>, KafkaPartitionSplit>
splitFetcher,
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
+ Set<TopicPartition> partitionsWithoutOffset,
OffsetCommitCallback callback) {
KafkaPartitionSplitReader kafkaReader =
(KafkaPartitionSplitReader) splitFetcher.getSplitReader();
@@ -101,7 +110,8 @@ public class KafkaSourceFetcherManager
new SplitFetcherTask() {
@Override
public boolean run() throws IOException {
- kafkaReader.notifyCheckpointComplete(offsetsToCommit,
callback);
+ kafkaReader.notifyCheckpointComplete(
+ offsetsToCommit, partitionsWithoutOffset,
callback);
return true;
}
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
index 95502959..0b5e5335 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
@@ -37,8 +37,11 @@ import org.apache.flink.runtime.metrics.MetricNames;
import org.apache.flink.runtime.metrics.groups.InternalSourceReaderMetricGroup;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
@@ -54,6 +57,7 @@ import org.junit.jupiter.params.provider.NullAndEmptySource;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
@@ -66,11 +70,16 @@ import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
import static
org.apache.flink.connector.kafka.testutils.KafkaSourceTestEnv.NUM_RECORDS_PER_PARTITION;
+import static org.apache.flink.core.testutils.CommonTestUtils.waitUtil;
+import static
org.apache.flink.streaming.connectors.kafka.KafkaTestBase.kafkaServer;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.entry;
/** Unit tests for {@link KafkaPartitionSplitReader}. */
@ResourceLock("KafkaTestBase")
@@ -425,6 +434,223 @@ public class KafkaPartitionSplitReaderTest {
KafkaSourceOptions.POLL_TIMEOUT_MS.key()));
}
+ @Test
+ void testTrackedOffsetsAreRemovedWhenPartitionsAreUnassigned() throws
Exception {
+ KafkaPartitionSplitReader reader = createReaderForCommits();
+ final TopicPartition finishingPartition = new TopicPartition(TOPIC1,
0);
+ // a second partition is kept assigned throughout, which prevents
consumer.poll()
+ // blocking, and proves that only offsets of the unassigned partition
are removed
+ final TopicPartition remainingPartition = new TopicPartition(TOPIC1,
1);
+ reader.handleSplitsChanges(
+ new SplitsAddition<>(
+ Arrays.asList(
+ new KafkaPartitionSplit(
+ finishingPartition,
+
earliestOffsets.get(finishingPartition),
+ NUM_RECORDS_PER_PARTITION),
+ new KafkaPartitionSplit(
+ remainingPartition,
+
earliestOffsets.get(remainingPartition),
+
KafkaPartitionSplit.NO_STOPPING_OFFSET))));
+
+ // a commit while both partitions are assigned records their committed
offsets
+ final Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new
HashMap<>();
+ offsetsToCommit.put(
+ finishingPartition, new
OffsetAndMetadata(earliestOffsets.get(finishingPartition)));
+ offsetsToCommit.put(
+ remainingPartition, new
OffsetAndMetadata(earliestOffsets.get(remainingPartition)));
+ commitOffsets(reader, offsetsToCommit, Collections.emptySet());
+
+ // fetch until the bounded split has been finished and unassigned
+ fetchUntil(
+ reader,
+ () ->
!reader.consumer().assignment().contains(finishingPartition),
+ "The bounded split was not finished.");
+
assertThat(reader.consumer().assignment()).contains(remainingPartition);
+
+ assertThat(reader.lastFetchedOffsets)
+ .as("Offsets tracked for committing")
+ .doesNotContainKey(finishingPartition)
+ .containsKey(remainingPartition);
+ assertThat(reader.lastKnownPositions)
+ .as("Consumer positions tracked for committing")
+ .doesNotContainKey(finishingPartition)
+ .containsKey(remainingPartition);
+ assertThat(reader.lastCommittedOffsets)
+ .as("Offsets last committed")
+ .doesNotContainKey(finishingPartition)
+ .containsKey(remainingPartition);
+ }
+
+ @Test
+ void testPartitionWithoutOffsetIsCommittedAtConsumerPosition() throws
Exception {
+ final KafkaPartitionSplitReader reader = createReaderForCommits();
+ // an empty partition, so the split never emits a record
+ final TopicPartition tp = new TopicPartition(TOPIC3, 0);
+ assignSplit(reader, new KafkaPartitionSplit(tp,
KafkaPartitionSplit.EARLIEST_OFFSET));
+ reader.fetch();
+
+ assertThat(commitPartitionsWithoutOffset(reader, tp))
+ .containsExactly(entry(tp, new OffsetAndMetadata(0L)));
+ }
+
+ @Test
+ void testPartitionWithoutOffsetIsNotCommittedOnceRecordsAreFetched()
throws Exception {
+ final KafkaPartitionSplitReader reader = createReaderForCommits();
+ final TopicPartition tp = new TopicPartition(TOPIC1, 0);
+ assignSplit(reader, new KafkaPartitionSplit(tp,
KafkaPartitionSplit.EARLIEST_OFFSET));
+ fetchUntil(
+ reader,
+ () -> reader.lastFetchedOffsets.containsKey(tp),
+ "No records were fetched.");
+
+ // the checkpoint does not include the fetched records, so the
position would skip them
+ assertThat(commitPartitionsWithoutOffset(reader, tp)).isEmpty();
+ }
+
+ @Test
+ void testPartitionWithoutOffsetIsNotCommittedForBoundedSplit() throws
Exception {
+ final KafkaPartitionSplitReader reader = createReaderForCommits();
+ final TopicPartition tp = new TopicPartition(TOPIC3, 1);
+ assignSplit(reader, new KafkaPartitionSplit(tp,
KafkaPartitionSplit.EARLIEST_OFFSET, 5L));
+ reader.fetch();
+
+ assertThat(commitPartitionsWithoutOffset(reader, tp)).isEmpty();
+ }
+
+ /**
+ * Reproduces a checkpoint whose offset is followed by a commit marker
being committed twice:
+ * first while the partition is idle (so the offset progresses over the
marker), and then second
+ * after a new record has been fetched (which must not move the committed
offset back).
+ */
+ @Test
+ void
testCommittedOffsetDoesNotGoBackwardsWhenRecordIsFetchedBeforeCommit() throws
Throwable {
+ final String topic = "CommittedOffsetDoesNotGoBackwards";
+ final String groupId = topic + "Group";
+ final TopicPartition tp = new TopicPartition(topic, 0);
+ KafkaSourceTestEnv.createTestTopic(topic, 1, 1);
+
+ // offset 0 = record
+ // offset 1 = commit marker
+ produceRecordInTransaction(tp);
+
+ final Properties props = new Properties();
+ props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ props.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,
"read_committed");
+ props.setProperty(KafkaSourceOptions.POLL_TIMEOUT_MS.key(), "100");
+ final KafkaPartitionSplitReader reader =
+ createReader(props,
UnregisteredMetricsGroup.createSourceReaderMetricGroup());
+ assignSplit(reader, new KafkaPartitionSplit(tp, 0L));
+
+ // both checkpoints are taken after the record at offset 0 is emitted
+ final Map<TopicPartition, OffsetAndMetadata> snapshot =
+ Collections.singletonMap(tp, new OffsetAndMetadata(1L));
+
+ // checkpoint 1 is committed while the partition is idle after the
commit marker
+ fetchUntil(
+ reader,
+ () ->
Long.valueOf(2L).equals(reader.lastKnownPositions.get(tp)),
+ "The consumer did not move past the commit marker.");
+ commitOffsets(reader, snapshot, Collections.emptySet());
+ assertThat(getCommittedOffset(tp, groupId))
+ .as("The committed offset after checkpoint 1")
+ .isEqualTo(2L);
+
+ // offset 2 = record
+ // offset 3 = commit marker
+ produceRecordInTransaction(tp);
+
+ // checkpoint 2 is committed after the record at offset 2 is fetched,
but not emitted
+ fetchUntil(
+ reader,
+ () ->
Long.valueOf(2L).equals(reader.lastFetchedOffsets.get(tp)),
+ "The record at offset 2 was not fetched.");
+ commitOffsets(reader, snapshot, Collections.emptySet());
+ assertThat(getCommittedOffset(tp, groupId))
+ .as("The committed offset after checkpoint 2")
+ .isEqualTo(2L);
+ }
+
+ /**
+ * The consumer only completes a commit on a later poll, so a commit for a
later checkpoint can
+ * be sent while an earlier one is still in flight.
+ */
+ @Test
+ void testCommitInFlightIsNotOvertakenByLowerOffset() throws Exception {
+ final String groupId = "CommitInFlightGroup";
+ final Properties props = new Properties();
+ props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ props.setProperty(KafkaSourceOptions.POLL_TIMEOUT_MS.key(), "100");
+ final KafkaPartitionSplitReader reader =
+ createReader(props,
UnregisteredMetricsGroup.createSourceReaderMetricGroup());
+ final TopicPartition tp = new TopicPartition(TOPIC1, 0);
+ // not fetched, so there is no consumer position to reconcile the
offsets with
+ assignSplit(reader, new KafkaPartitionSplit(tp,
KafkaPartitionSplit.EARLIEST_OFFSET));
+
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> first =
+ startCommit(
+ reader,
+ Collections.singletonMap(tp, new
OffsetAndMetadata(5L)),
+ Collections.emptySet());
+ assertThat(first).as("The first commit is still in
flight").isNotDone();
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> second
=
+ startCommit(
+ reader,
+ Collections.singletonMap(tp, new
OffsetAndMetadata(3L)),
+ Collections.emptySet());
+ fetchUntil(
+ reader,
+ () -> first.isDone() && second.isDone(),
+ "The offset commits did not complete.");
+
+ assertThat(first.get()).containsExactly(entry(tp, new
OffsetAndMetadata(5L)));
+ assertThat(second.get()).isEmpty();
+ assertThat(getCommittedOffset(tp, groupId)).isEqualTo(5L);
+ }
+
+ /**
+ * The consumer would otherwise only run the callback on a later poll,
which on an idle
+ * partition is a poll timeout away.
+ */
+ @Test
+ void testCommitCompletesImmediatelyWhenThereIsNothingToCommit() {
+ final KafkaPartitionSplitReader reader = createReaderForCommits();
+ final TopicPartition tp = new TopicPartition(TOPIC3, 1);
+ // a bounded split is not committed at the consumer position, so
nothing is committed
+ assignSplit(reader, new KafkaPartitionSplit(tp,
KafkaPartitionSplit.EARLIEST_OFFSET, 5L));
+
+ assertThat(startCommit(reader, Collections.emptyMap(),
Collections.singleton(tp)))
+ .isCompletedWithValue(Collections.emptyMap());
+ }
+
+ /**
+ * Partitions are only unassigned when their split finishes, which also
removes their tracked
+ * consumer position. A stale position is injected here, so that the
assignment is checked on
+ * its own.
+ */
+ @Test
+ void testOffsetsOfUnassignedPartitionsAreNotReconciled() throws Exception {
+ final KafkaPartitionSplitReader reader = createReaderForCommits();
+ // keeps the consumer assigned, so that it can poll for the commit to
complete
+ assignSplit(
+ reader,
+ new KafkaPartitionSplit(
+ new TopicPartition(TOPIC3, 0),
KafkaPartitionSplit.EARLIEST_OFFSET));
+ final TopicPartition unassignedWithOffset = new TopicPartition(TOPIC1,
0);
+ final TopicPartition unassignedWithoutOffset = new
TopicPartition(TOPIC1, 1);
+ reader.lastKnownPositions.put(unassignedWithOffset, 5L);
+ reader.lastKnownPositions.put(unassignedWithoutOffset, 5L);
+
+ assertThat(
+ commitOffsets(
+ reader,
+ Collections.singletonMap(
+ unassignedWithOffset, new
OffsetAndMetadata(2L)),
+
Collections.singleton(unassignedWithoutOffset)))
+ .containsExactly(entry(unassignedWithOffset, new
OffsetAndMetadata(2L)));
+
assertThat(reader.lastCommittedOffsets).doesNotContainKey(unassignedWithOffset);
+ }
+
// ------------------
private void assignSplitsAndFetchUntilFinish(KafkaPartitionSplitReader
reader, int readerId)
@@ -511,6 +737,95 @@ public class KafkaPartitionSplitReaderTest {
rackId);
}
+ private KafkaPartitionSplitReader createReaderForCommits() {
+ final Properties props = new Properties();
+ props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,
"PartitionWithoutOffsetCommitGroup");
+ // the partitions under test have no records to deliver, so do not
wait for them
+ props.setProperty(KafkaSourceOptions.POLL_TIMEOUT_MS.key(), "100");
+ return createReader(props,
UnregisteredMetricsGroup.createSourceReaderMetricGroup());
+ }
+
+ private static void assignSplit(KafkaPartitionSplitReader reader,
KafkaPartitionSplit split) {
+ reader.handleSplitsChanges(new
SplitsAddition<>(Collections.singletonList(split)));
+ }
+
+ private static Map<TopicPartition, OffsetAndMetadata>
commitPartitionsWithoutOffset(
+ KafkaPartitionSplitReader reader, TopicPartition tp) throws
Exception {
+ return commitOffsets(reader, Collections.emptyMap(),
Collections.singleton(tp));
+ }
+
+ /** Returns the offsets that the reader committed to Kafka for a completed
checkpoint. */
+ private static Map<TopicPartition, OffsetAndMetadata> commitOffsets(
+ KafkaPartitionSplitReader reader,
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
+ Set<TopicPartition> partitionsWithoutOffset)
+ throws Exception {
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>
committed =
+ startCommit(reader, offsetsToCommit, partitionsWithoutOffset);
+ // the consumer only invokes the callback of a commit sent to Kafka on
a later poll
+ fetchUntil(reader, committed::isDone, "The offset commit did not
complete.");
+ return committed.get();
+ }
+
+ /** Returns the offsets that the reader commits to Kafka, without polling
for them. */
+ private static CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>
startCommit(
+ KafkaPartitionSplitReader reader,
+ Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
+ Set<TopicPartition> partitionsWithoutOffset) {
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>
committed =
+ new CompletableFuture<>();
+ reader.notifyCheckpointComplete(
+ offsetsToCommit,
+ partitionsWithoutOffset,
+ (offsets, e) -> {
+ if (e != null) {
+ committed.completeExceptionally(e);
+ } else {
+ committed.complete(offsets);
+ }
+ });
+ return committed;
+ }
+
+ private static void fetchUntil(
+ KafkaPartitionSplitReader reader, Supplier<Boolean> condition,
String errorMsg)
+ throws Exception {
+ waitUtil(
+ () -> {
+ if (!condition.get()) {
+ try {
+ reader.fetch();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+ return condition.get();
+ },
+ Duration.ofSeconds(30),
+ errorMsg);
+ }
+
+ private static long getCommittedOffset(TopicPartition tp, String groupId)
throws Exception {
+ try (AdminClient adminClient = KafkaSourceTestEnv.getAdminClient()) {
+ final OffsetAndMetadata committed =
+ adminClient
+ .listConsumerGroupOffsets(groupId)
+ .partitionsToOffsetAndMetadata()
+ .get()
+ .get(tp);
+ assertThat(committed).as("No offset was committed for %s",
tp).isNotNull();
+ return committed.offset();
+ }
+ }
+
+ /** Writes a transaction with a single record, which takes two offsets
with its marker. */
+ private static void produceRecordInTransaction(TopicPartition tp) throws
Throwable {
+ KafkaSourceTestEnv.produceToKafka(
+ Collections.singletonList(
+ new ProducerRecord<>(tp.topic(), tp.partition(),
tp.toString(), 0)),
+ kafkaServer.getTransactionalProducerConfig());
+ }
+
private Map<String, KafkaPartitionSplit> assignSplits(
KafkaPartitionSplitReader reader, Map<String, KafkaPartitionSplit>
splits) {
SplitsChange<KafkaPartitionSplit> splitsChange =
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
index 9b5221f6..f756ef4a 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
@@ -43,17 +43,27 @@ import
org.apache.flink.runtime.metrics.groups.InternalSourceReaderMetricGroup;
import org.apache.flink.util.function.SerializableSupplier;
import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.RecordsToDelete;
import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.parallel.ResourceLock;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import javax.annotation.Nullable;
import java.time.Duration;
import java.util.ArrayList;
@@ -67,9 +77,12 @@ import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Supplier;
+import java.util.stream.Stream;
import static
org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.COMMITS_SUCCEEDED_METRIC_COUNTER;
import static
org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.COMMITTED_OFFSET_METRIC_GAUGE;
@@ -81,7 +94,9 @@ import static
org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderM
import static
org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.TOPIC_GROUP;
import static
org.apache.flink.connector.kafka.testutils.KafkaSourceTestEnv.NUM_PARTITIONS;
import static org.apache.flink.core.testutils.CommonTestUtils.waitUtil;
+import static
org.apache.flink.streaming.connectors.kafka.KafkaTestBase.kafkaServer;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.entry;
/** Unit tests for {@link KafkaSourceReader}. */
@ResourceLock("KafkaTestBase")
@@ -240,6 +255,7 @@ public class KafkaSourceReaderTest extends
SourceReaderTestBase<KafkaPartitionSp
() -> {
try {
reader.notifyCheckpointComplete(lastCheckpointId);
+ reader.pollNext(output);
} catch (Exception exception) {
throw new RuntimeException(
"Caught unexpected exception when polling
from the reader",
@@ -266,17 +282,635 @@ public class KafkaSourceReaderTest extends
SourceReaderTestBase<KafkaPartitionSp
}
}
+ /** Writes the records that a {@link #offsetConvergenceScenarios} scenario
needs. */
+ @FunctionalInterface
+ private interface RecordProducer {
+ void produce(String topic) throws Throwable;
+ }
+
+ private static Stream<Arguments> offsetConvergenceScenarios() {
+ return Stream.of(
+ // no transaction and just one record
+ // offset 0 = record
+ // offset 1 << log end offset
+ Arguments.of(
+ "NonTransactional",
+ 0L,
+ 1L,
+ 1,
+ 1,
+ (RecordProducer)
+ topic ->
+ KafkaSourceTestEnv.produceToKafka(
+ Collections.singletonList(
+ new ProducerRecord<>(
+ topic, 0,
topic + "-key", 0)))),
+ // transaction containing one record and a Kafka commit marker
+ // offset 0 = record
+ // offset 1 = commit marker
+ // offset 2 << log end offset
+ Arguments.of(
+ "Transactional",
+ 0L,
+ 2L,
+ 1,
+ 1,
+ (RecordProducer)
+ topic ->
+ KafkaSourceTestEnv.produceToKafka(
+ Collections.singletonList(
+ new ProducerRecord<>(
+ topic, 0,
topic + "-key", 0)),
+
kafkaServer.getTransactionalProducerConfig())),
+ // transaction containing three records
+ // offset 0 = record
+ // offset 1 = record
+ // offset 2 = record
+ // offset 3 = commit marker
+ // offset 4 << log end offset
+ Arguments.of(
+ "MultiRecordTransaction",
+ 0L,
+ 4L,
+ 3,
+ 1,
+ (RecordProducer) topic -> produceInTransaction(topic,
0, 3, true)),
+ // committed transaction containing two records
+ // aborted transaction containing three records
+ // offset 0 = record
+ // offset 1 = record
+ // offset 2 = commit marker
+ // offset 3 = record
+ // offset 4 = record
+ // offset 5 = record
+ // offset 6 = abort marker
+ // offset 7 << log end offset
+ Arguments.of(
+ "AbortedTransaction",
+ 0L,
+ 7L,
+ 2,
+ 1,
+ (RecordProducer)
+ topic -> {
+ produceInTransaction(topic, 0, 2, true);
+ produceInTransaction(topic, 0, 3, false);
+ }),
+ // three separate transactions each containing one record
+ // offset 0 = record
+ // offset 1 = commit marker
+ // offset 2 = record
+ // offset 3 = commit marker
+ // offset 4 = record
+ // offset 5 = commit marker
+ // offset 6 << log end offset
+ Arguments.of(
+ "MultipleTransactions",
+ 0L,
+ 6L,
+ 3,
+ 1,
+ (RecordProducer)
+ topic -> {
+ produceInTransaction(topic, 0, 1, true);
+ produceInTransaction(topic, 0, 1, true);
+ produceInTransaction(topic, 0, 1, true);
+ }),
+ // two transactional producers interleaving records on the
same partition before
+ // either commits, so the two commit markers do not
immediately follow their own
+ // records
+ // offset 0 = producer A record
+ // offset 1 = producer B record
+ // offset 2 = producer A record
+ // offset 3 = producer B record
+ // offset 4 = producer A commit marker
+ // offset 5 = producer B commit marker
+ // offset 6 << log end offset
+ Arguments.of(
+ "InterleavedTransactions",
+ 0L,
+ 6L,
+ 4,
+ 1,
+ (RecordProducer) topic ->
produceInterleavedTransactions(topic, 0, 2)),
+ // a single aborted transaction
+ // so the partition never delivers a record to the reader
+ // offset 0 = record
+ // offset 1 = record
+ // offset 2 = record
+ // offset 3 = abort marker
+ // offset 4 << log end offset
+ Arguments.of(
+ "OnlyAbortedTransaction",
+ 0L,
+ 4L,
+ 0,
+ 2,
+ (RecordProducer) topic -> produceInTransaction(topic,
0, 3, false)),
+ // the same single aborted transaction, read from the earliest
offset as
+ // OffsetsInitializer.earliest() does, so the split never
holds a real offset
+ Arguments.of(
+ "OnlyAbortedTransactionFromEarliest",
+ KafkaPartitionSplit.EARLIEST_OFFSET,
+ 4L,
+ 0,
+ 2,
+ (RecordProducer) topic -> produceInTransaction(topic,
0, 3, false)));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("offsetConvergenceScenarios")
+ void testCommittedOffsetConvergesWithLogEndOffset(
+ String scenario,
+ long startingOffset,
+ long expectedLastStableOffset,
+ int expectedReadableRecords,
+ int maxCheckpoints,
+ RecordProducer recordProducer)
+ throws Throwable {
+ final String topic = TOPIC + scenario;
+ final String groupId = scenario + "OffsetCommitGroup";
+ final TopicPartition tp = new TopicPartition(topic, 0);
+ KafkaSourceTestEnv.createTestTopic(topic, 1, 1);
+
+ recordProducer.produce(topic);
+
+ // assert state of Kafka
+ final long lastStableOffset = awaitLastStableOffset(tp,
expectedLastStableOffset);
+ assertThat(getReadCommittedRecordsCount(tp))
+ .as("Number of readable records")
+ .isEqualTo(expectedReadableRecords);
+
+ // run KafkaSourceReader and check the last offset committed
+ final long committedOffset =
+ readAndCommitOffset(
+ tp,
+ startingOffset,
+ groupId,
+ expectedReadableRecords,
+ maxCheckpoints,
+ lastStableOffset,
+ new Properties());
+ assertThat(committedOffset)
+ .as("The committed offset should converge with the log end
offset")
+ .isEqualTo(lastStableOffset);
+ }
+
+ @Test
+ void testCommittedOffsetDoesNotOverrunRecordsInATransaction() throws
Throwable {
+ final String topic = TOPIC + "PartialTransaction";
+ final String groupId = "PartialTransactionOffsetCommitGroup";
+ final TopicPartition tp = new TopicPartition(topic, 0);
+ KafkaSourceTestEnv.createTestTopic(topic, 1, 1);
+
+ // transaction containing three records
+ // offset 0 = record
+ // offset 1 = record
+ // offset 2 = record
+ // offset 3 = commit marker
+ // offset 4 << log end offset
+ produceInTransaction(topic, 0, 3, true);
+
+ // assert state of Kafka
+ awaitLastStableOffset(tp, 4L);
+ final int numReadableOffsets = getReadCommittedRecordsCount(tp);
+ assertThat(numReadableOffsets).as("Three records should be
readable").isEqualTo(3);
+
+ // run KafkaSourceReader and check the last offset committed
+ final int recordsToRead = 2;
+ final long committedOffset =
+ readAndCommitOffset(tp, groupId, recordsToRead, 1,
Long.MIN_VALUE);
+ assertThat(committedOffset)
+ .as("The committed offset should be the next offset after the
two read records")
+ .isEqualTo(2);
+ }
+
+ /**
+ * Simulates a low-traffic bursty transactional topic, where a burst of
records are followed by
+ * a commit marker. The result of this could be for a commit marker to be
returned with no
+ * records in a poll. The intent of this test is to verify that the offset
converges in such a
+ * scenario, and does not get stuck with an incorrect lag until more
records arrive on the topic
+ * to allow an update.
+ */
+ @Test
+ void
testCommittedOffsetConvergesOnIdlePartitionAfterTrailingCommitMarker() throws
Throwable {
+ final String topic = TOPIC + "IdleAfterTrailingMarker";
+ final String groupId = "IdleAfterTrailingMarkerOffsetCommitGroup";
+ final TopicPartition tp = new TopicPartition(topic, 0);
+ KafkaSourceTestEnv.createTestTopic(topic, 1, 1);
+
+ // transaction containing two records
+ // offset 0 = record
+ // offset 1 = record
+ // offset 2 = commit marker
+ // offset 3 << log end offset
+ produceInTransaction(topic, 0, 2, true);
+
+ // assert state of Kafka
+ final long lastStableOffset = awaitLastStableOffset(tp, 3L);
+ assertThat(getReadCommittedRecordsCount(tp)).as("Number of readable
records").isEqualTo(2);
+
+ // one record per poll, so that the commit marker is left to a poll of
its own
+ final Properties oneRecordPerPoll = new Properties();
+ oneRecordPerPoll.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,
"1");
+
+ // the poll that skips the commit marker is not the poll that delivers
the last
+ // record, so allow for the first checkpoint being completed in
between the two
+ final int maxCheckpoints = 2;
+
+ // run KafkaSourceReader and check the last offset committed
+ final long committedOffset =
+ readAndCommitOffset(
+ tp, groupId, 2, maxCheckpoints, lastStableOffset,
oneRecordPerPoll);
+ assertThat(committedOffset)
+ .as("The committed offset should converge with the log end
offset")
+ .isEqualTo(lastStableOffset);
+ }
+
+ @SuppressWarnings("deprecation")
+ private static Stream<Arguments> emptyBoundedSplitScenarios() {
+ return Stream.of(
+ // the group has already committed past the split's stopping
offset
+ Arguments.of(
+ "CommittedStart",
KafkaPartitionSplit.COMMITTED_OFFSET, 5L, 8L, null, 8L),
+ // retention has already removed the records before the
split's stopping offset
+ Arguments.of(
+ "EarliestStartAfterRetention",
+ KafkaPartitionSplit.EARLIEST_OFFSET,
+ 3L,
+ null,
+ 6L,
+ null),
+ // a split restored from older state, which starts from the
latest offset
+ Arguments.of("LatestStart", KafkaPartitionSplit.LATEST_OFFSET,
5L, 8L, null, 8L),
+ // a split that starts from a concrete offset past its
stopping offset
+ Arguments.of("ConcreteStart", 7L, 5L, null, null, 7L));
+ }
+
+ /**
+ * A split that is empty when it is assigned, because the consumer starts
at or past its
+ * stopping offset, is finished without reading anything. Its stopping
offset must not be
+ * committed, as the consumer never read up to it.
+ */
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("emptyBoundedSplitScenarios")
+ void testFinishedEmptySplitDoesNotCommitItsStoppingOffset(
+ String scenario,
+ long startingOffset,
+ long stoppingOffset,
+ @Nullable Long committedOffsetBefore,
+ @Nullable Long logStartOffset,
+ @Nullable Long expectedCommittedOffset)
+ throws Throwable {
+ final String topic = TOPIC + "EmptySplit" + scenario;
+ final String groupId = scenario + "EmptySplitOffsetCommitGroup";
+ final TopicPartition tp = new TopicPartition(topic, 0);
+ // second partition to keep the fetcher polling after the empty split
finishes, so that
+ // the consumer completes the offset commit
+ final TopicPartition idleTp = new TopicPartition(topic, 1);
+ KafkaSourceTestEnv.createTestTopic(topic, 2, 1);
+
+ // offsets 0 - 9 = records
+ // offset 10 = log end offset
+ final List<ProducerRecord<String, Integer>> records = new
ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ records.add(new ProducerRecord<>(topic, 0, topic + "-key", i));
+ }
+ KafkaSourceTestEnv.produceToKafka(records);
+ try (AdminClient adminClient = KafkaSourceTestEnv.getAdminClient()) {
+ if (committedOffsetBefore != null) {
+ adminClient
+ .alterConsumerGroupOffsets(
+ groupId,
+ Collections.singletonMap(
+ tp, new
OffsetAndMetadata(committedOffsetBefore)))
+ .all()
+ .get();
+ }
+ if (logStartOffset != null) {
+ adminClient
+ .deleteRecords(
+ Collections.singletonMap(
+ tp,
RecordsToDelete.beforeOffset(logStartOffset)))
+ .all()
+ .get();
+ }
+ }
+
+ final Properties props = new Properties();
+ props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ try (KafkaSourceReader<Integer> reader =
+ (KafkaSourceReader<Integer>)
+ createReader(
+ Boundedness.CONTINUOUS_UNBOUNDED,
+ new TestingReaderContext(),
+ (ignore) -> {},
+ props,
+ null)) {
+ reader.addSplits(
+ Arrays.asList(
+ new KafkaPartitionSplit(tp, startingOffset,
stoppingOffset),
+ new KafkaPartitionSplit(
+ idleTp, 0L,
KafkaPartitionSplit.NO_STOPPING_OFFSET)));
+
+ final TestingReaderOutput<Integer> output = new
TestingReaderOutput<>();
+ pollUntilSplitFinished(reader, output, tp);
+ assertThat(output.getEmittedRecords()).isEmpty();
+
+ completeCheckpoint(reader, output, 1L);
+ assertThat(
+ Optional.ofNullable(findCommittedOffset(tp,
groupId))
+ .map(OffsetAndMetadata::offset))
+ .as("The offset committed for %s", tp)
+ .isEqualTo(Optional.ofNullable(expectedCommittedOffset));
+ }
+ }
+
+ @Test
+ void testCommittedOffsetForBoundedSplitDoesNotGoBackwards() throws
Throwable {
+ final String topic = TOPIC + "BoundedTrailingMarkers";
+ final String groupId = "BoundedTrailingMarkersOffsetCommitGroup";
+ final TopicPartition tp = new TopicPartition(topic, 0);
+ // second partition to keep the fetcher polling after the bounded
split finishes
+ final TopicPartition idleTp = new TopicPartition(topic, 1);
+ KafkaSourceTestEnv.createTestTopic(topic, 2, 1);
+
+ // committed transaction containing two records
+ // offset 0 = record
+ // offset 1 = record
+ // offset 2 = commit marker
+ // offset 3 << log end offset
+ produceInTransaction(topic, 0, 2, true);
+ awaitLastStableOffset(tp, 3L);
+
+ final Properties props = new Properties();
+ props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ props.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,
"read_committed");
+
+ try (KafkaSourceReader<Integer> reader =
+ (KafkaSourceReader<Integer>)
+ createReader(
+ Boundedness.CONTINUOUS_UNBOUNDED,
+ new TestingReaderContext(),
+ (ignore) -> {},
+ props,
+ null)) {
+ reader.addSplits(
+ Arrays.asList(
+ new KafkaPartitionSplit(tp, 0L, 10L),
+ new KafkaPartitionSplit(
+ idleTp, 0L,
KafkaPartitionSplit.NO_STOPPING_OFFSET)));
+
+ final TestingReaderOutput<Integer> output = new
TestingReaderOutput<>();
+ pollUntil(
+ reader,
+ output,
+ () -> output.getEmittedRecords().size() == 2,
+ "The reader did not emit all records before timeout.");
+
+ completeCheckpoint(reader, output, 1L);
+ final long offsetWhileRunning = getCommittedOffset(tp, groupId);
+ assertThat(offsetWhileRunning)
+ .as(
+ "The committed offset of a running bounded split,
which is not advanced"
+ + " over the trailing commit marker")
+ .isEqualTo(2L);
+
+ // aborted transaction that takes the partition up to the split's
stopping offset
+ // offsets 3 - 8 = records
+ // offset 9 = abort marker
+ // offset 10 = log end offset
+ produceInTransaction(topic, 0, 6, false);
+ awaitLastStableOffset(tp, 10L);
+ pollUntilSplitFinished(reader, output, tp);
+
+ completeCheckpoint(reader, output, 2L);
+ final long offsetAfterFinishing = getCommittedOffset(tp, groupId);
+
+ assertThat(offsetAfterFinishing)
+ .as("The committed offset of a finished bounded split is
its stopping offset")
+ .isEqualTo(10L);
+ }
+ }
+
+ /**
+ * Snapshots the reader and waits for the resulting offsets to reach
Kafka, polling the reader
+ * throughout, for consistency with how the mailbox thread interleaves
pollNext with checkpoint
+ * callbacks.
+ */
+ private void completeCheckpoint(
+ KafkaSourceReader<Integer> reader, ReaderOutput<Integer> output,
long checkpointId)
+ throws Exception {
+ reader.snapshotState(checkpointId);
+ waitUtil(
+ () -> {
+ try {
+ reader.notifyCheckpointComplete(checkpointId);
+ reader.pollNext(output);
+ } catch (Exception exception) {
+ throw new RuntimeException(
+ "Unexpected exception while committing",
exception);
+ }
+ return reader.getOffsetsToCommit().isEmpty();
+ },
+ Duration.ofSeconds(30),
+ Duration.ofMillis(500),
+ "Offset commit did not finish before timeout.");
+ }
+
+ /**
+ * Reads {@code expectedRecords} records from {@code tp} using a
read_committed reader, and then
+ * completes checkpoints while the partition is idle until either the
committed offset reaches
+ * {@code targetOffset} or {@code maxCheckpoints} have been completed.
Returns the last offset
+ * that the reader committed to Kafka.
+ */
+ private long readAndCommitOffset(
+ TopicPartition tp,
+ String groupId,
+ int expectedRecords,
+ int maxCheckpoints,
+ long targetOffset)
+ throws Exception {
+ return readAndCommitOffset(
+ tp, groupId, expectedRecords, maxCheckpoints, targetOffset,
new Properties());
+ }
+
+ private long readAndCommitOffset(
+ TopicPartition tp,
+ String groupId,
+ int expectedRecords,
+ int maxCheckpoints,
+ long targetOffset,
+ Properties extraConsumerProps)
+ throws Exception {
+ return readAndCommitOffset(
+ tp, 0L, groupId, expectedRecords, maxCheckpoints,
targetOffset, extraConsumerProps);
+ }
+
+ private long readAndCommitOffset(
+ TopicPartition tp,
+ long startingOffset,
+ String groupId,
+ int expectedRecords,
+ int maxCheckpoints,
+ long targetOffset,
+ Properties extraConsumerProps)
+ throws Exception {
+ final Properties props = new Properties();
+ props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ props.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,
"read_committed");
+ props.putAll(extraConsumerProps);
+
+ final MetricListener metricListener = new MetricListener();
+
+ try (KafkaSourceReader<Integer> reader =
+ (KafkaSourceReader<Integer>)
+ createReader(
+ Boundedness.CONTINUOUS_UNBOUNDED,
+ new TestingReaderContext(
+ new Configuration(),
+ InternalSourceReaderMetricGroup.mock(
+
metricListener.getMetricGroup())),
+ (ignore) -> {},
+ props,
+ null)) {
+ // prepare kafka reader
+ reader.addSplits(
+ Collections.singletonList(
+ new KafkaPartitionSplit(
+ tp, startingOffset,
KafkaPartitionSplit.NO_STOPPING_OFFSET)));
+
+ // read until partition is idle
+ final TestingReaderOutput<Integer> output = new
TestingReaderOutput<>();
+ pollUntil(
+ reader,
+ output,
+ () -> output.getEmittedRecords().size() == expectedRecords,
+ "The reader did not emit all records before timeout.");
+
+ // complete checkpoints to commit offset
+ long committedOffset = Long.MIN_VALUE;
+ for (int checkpointId = 1; checkpointId <= maxCheckpoints;
checkpointId++) {
+ if (checkpointId > 1) {
+ reader.pollNext(output);
+ }
+ completeCheckpoint(reader, output, checkpointId);
+ final OffsetAndMetadata committed = findCommittedOffset(tp,
groupId);
+ if (committed == null && checkpointId < maxCheckpoints) {
+ // a split that has not emitted a record is committed at
the consumer
+ // position, which is unknown until the split reader has
polled the partition
+ continue;
+ }
+ assertThat(committed).as("No offset was committed for %s",
tp).isNotNull();
+ committedOffset = committed.offset();
+ assertThat(getCommittedOffsetMetric(tp, metricListener))
+ .as("The committedOffsets metric should match the
offset held by Kafka")
+ .isEqualTo(committedOffset);
+ if (committedOffset == targetOffset) {
+ break;
+ }
+ }
+ return committedOffset;
+ }
+ }
+
+ /**
+ * Writes {@code numRecords} records to a partition within a single
transaction, and then
+ * commits or aborts the transaction before closing the producer.
+ */
+ private static void produceInTransaction(
+ String topic, int partition, int numRecords, boolean commit) {
+ try (KafkaProducer<String, Integer> producer =
createTransactionalProducer()) {
+ producer.initTransactions();
+ producer.beginTransaction();
+ for (int i = 0; i < numRecords; i++) {
+ producer.send(new ProducerRecord<>(topic, partition, topic +
"-key", i));
+ }
+ producer.flush();
+ if (commit) {
+ producer.commitTransaction();
+ } else {
+ producer.abortTransaction();
+ }
+ }
+ }
+
+ private static void produceInterleavedTransactions(
+ String topic, int partition, int numRecordsPerProducer) throws
Exception {
+ try (KafkaProducer<String, Integer> producerA =
createTransactionalProducer();
+ KafkaProducer<String, Integer> producerB =
createTransactionalProducer()) {
+ producerA.initTransactions();
+ producerB.initTransactions();
+ producerA.beginTransaction();
+ producerB.beginTransaction();
+ for (int i = 0; i < numRecordsPerProducer; i++) {
+ producerA.send(new ProducerRecord<>(topic, partition, topic +
"-key", i)).get();
+ producerB
+ .send(new ProducerRecord<>(topic, partition, topic +
"-key", 100 + i))
+ .get();
+ }
+ producerA.commitTransaction();
+ producerB.commitTransaction();
+ }
+ }
+
+ private static KafkaProducer<String, Integer>
createTransactionalProducer() {
+ final Properties props = new Properties();
+ props.putAll(KafkaSourceTestEnv.standardProps);
+ props.putAll(kafkaServer.getIdempotentProducerConfig());
+ props.putAll(kafkaServer.getTransactionalProducerConfig());
+ props.setProperty(
+ ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
+ props.setProperty(
+ ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
IntegerSerializer.class.getName());
+ return new KafkaProducer<>(props);
+ }
+
@Test
void testNotCommitOffsetsForUninitializedSplits() throws Exception {
final long checkpointId = 1234L;
try (KafkaSourceReader<Integer> reader = (KafkaSourceReader<Integer>)
createReader()) {
+ final TopicPartition tp = new TopicPartition(TOPIC, 0);
KafkaPartitionSplit split =
- new KafkaPartitionSplit(
- new TopicPartition(TOPIC, 0),
KafkaPartitionSplit.EARLIEST_OFFSET);
+ new KafkaPartitionSplit(tp,
KafkaPartitionSplit.EARLIEST_OFFSET);
reader.addSplits(Collections.singletonList(split));
reader.snapshotState(checkpointId);
assertThat(reader.getOffsetsToCommit()).hasSize(1);
assertThat(reader.getOffsetsToCommit().get(checkpointId)).isEmpty();
+ // the split reader commits the consumer position for it instead
+ assertThat(reader.getPartitionsWithoutOffsetToCommit())
+ .containsExactly(entry(checkpointId,
Collections.singleton(tp)));
+ }
+ }
+
+ @Test
+ void testNoCommitIsCountedWhenThereIsNothingToCommit() throws Exception {
+ final MetricListener metricListener = new MetricListener();
+ final String groupId = "NothingToCommitGroup";
+ final TopicPartition tp = new TopicPartition(TOPIC, 0);
+ try (KafkaSourceReader<Integer> reader =
+ (KafkaSourceReader<Integer>)
+ createReader(
+ Boundedness.CONTINUOUS_UNBOUNDED,
+ groupId,
+ metricListener.getMetricGroup())) {
+ // the checkpoint is taken before the split emits a record, and a
bounded split is not
+ // committed at the consumer position, so the commit has nothing
in it
+ reader.addSplits(
+ Collections.singletonList(
+ new KafkaPartitionSplit(
+ tp,
+ KafkaPartitionSplit.EARLIEST_OFFSET,
+ NUM_RECORDS_PER_SPLIT)));
+ completeCheckpoint(reader, new TestingReaderOutput<>(), 1L);
+
+ assertThat(findCommittedOffset(tp, groupId)).isNull();
+ assertThat(
+ metricListener.getCounter(
+ KAFKA_SOURCE_READER_METRIC_GROUP,
+ COMMITS_SUCCEEDED_METRIC_COUNTER))
+ .hasValueSatisfying(counter ->
assertThat(counter.getCount()).isZero());
+ assertThat(reader.getPartitionsWithoutOffsetToCommit()).isEmpty();
}
}
@@ -356,6 +990,7 @@ public class KafkaSourceReaderTest extends
SourceReaderTestBase<KafkaPartitionSp
() -> {
try {
reader.notifyCheckpointComplete(checkpointId);
+ reader.pollNext(output);
} catch (Exception e) {
throw new RuntimeException(
"Failed to notify checkpoint complete to
reader", e);
@@ -626,6 +1261,9 @@ public class KafkaSourceReaderTest extends
SourceReaderTestBase<KafkaPartitionSp
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
KafkaSourceTestEnv.brokerConnectionStrings)
.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
"false")
+ // Offset commits run between polls, so a short poll
timeout keeps each
+ // commit on an idle split from introducing a ~10sec
test pause
+ .setProperty(KafkaSourceOptions.POLL_TIMEOUT_MS.key(),
"100")
.setProperties(props);
if (boundedness == Boundedness.BOUNDED) {
builder.setBounded(OffsetsInitializer.latest());
@@ -659,6 +1297,23 @@ public class KafkaSourceReaderTest extends
SourceReaderTestBase<KafkaPartitionSp
errorMessage);
}
+ /**
+ * Polls the reader until it has handled the finishing of the split for
{@code tp}. The
+ * split-finished hook is not enough, as it fires once the finished split
has been queued for
+ * the reader, so this probes the reader's own state with a snapshot that
is never committed.
+ */
+ private void pollUntilSplitFinished(
+ KafkaSourceReader<Integer> reader, ReaderOutput<Integer> output,
TopicPartition tp)
+ throws Exception {
+ pollUntil(
+ reader,
+ output,
+ () ->
+ reader.snapshotState(0L).stream()
+ .noneMatch(split ->
split.getTopicPartition().equals(tp)),
+ "The split did not finish before timeout.");
+ }
+
private long getKafkaConsumerMetric(String name, MetricListener listener) {
final Optional<Gauge<Object>> kafkaConsumerGauge =
listener.getGauge(
@@ -695,6 +1350,76 @@ public class KafkaSourceReaderTest extends
SourceReaderTestBase<KafkaPartitionSp
// ---------------------
+ private static KafkaConsumer<String, String> createReadCommittedProbe() {
+ final Properties props = new Properties();
+ props.putAll(KafkaSourceTestEnv.standardProps);
+ props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "read-probe-" +
UUID.randomUUID());
+ props.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,
"read_committed");
+ props.setProperty(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ props.setProperty(
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ return new KafkaConsumer<>(props);
+ }
+
+ private static int getReadCommittedRecordsCount(TopicPartition tp) throws
Exception {
+ try (KafkaConsumer<String, String> probe = createReadCommittedProbe())
{
+ final List<TopicPartition> partitions =
Collections.singletonList(tp);
+ probe.assign(partitions);
+ probe.seekToBeginning(partitions);
+ final long lastStableOffset = probe.endOffsets(partitions).get(tp);
+ final AtomicInteger count = new AtomicInteger();
+ waitUtil(
+ () -> {
+
count.addAndGet(probe.poll(Duration.ofMillis(500)).records(tp).size());
+ return probe.position(tp) >= lastStableOffset;
+ },
+ Duration.ofSeconds(30),
+ String.format(
+ "The probe did not reach the last stable offset %d
of %s.",
+ lastStableOffset, tp));
+ return count.get();
+ }
+ }
+
+ /**
+ * Waits until the last stable offset of {@code tp} reaches {@code
expectedOffset}, and returns
+ * it. In most cases, this will return the expected value on the first
check, but the
+ * transaction coordinator propagates it to partition leaders
asynchronously, so LSO can briefly
+ * lag behind a committed transaction.
+ */
+ private static long awaitLastStableOffset(TopicPartition tp, long
expectedOffset)
+ throws Exception {
+ try (KafkaConsumer<String, String> probe = createReadCommittedProbe())
{
+ final List<TopicPartition> partitions =
Collections.singletonList(tp);
+ waitUtil(
+ () -> probe.endOffsets(partitions).get(tp) ==
expectedOffset,
+ Duration.ofSeconds(30),
+ Duration.ofMillis(200),
+ String.format(
+ "The last stable offset of %s did not reach %d.",
tp, expectedOffset));
+ return expectedOffset;
+ }
+ }
+
+ private static long getCommittedOffset(TopicPartition tp, String groupId)
throws Exception {
+ final OffsetAndMetadata committed = findCommittedOffset(tp, groupId);
+ assertThat(committed).as("No offset was committed for %s",
tp).isNotNull();
+ return committed.offset();
+ }
+
+ @Nullable
+ private static OffsetAndMetadata findCommittedOffset(TopicPartition tp,
String groupId)
+ throws Exception {
+ try (AdminClient adminClient = KafkaSourceTestEnv.getAdminClient()) {
+ return adminClient
+ .listConsumerGroupOffsets(groupId)
+ .partitionsToOffsetAndMetadata()
+ .get()
+ .get(tp);
+ }
+ }
+
private static List<ProducerRecord<String, Integer>> getRecords() {
List<ProducerRecord<String, Integer>> records = new ArrayList<>();
for (int part = 0; part < NUM_PARTITIONS; part++) {
