This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 0007e8b9e4a Use the empty-batch stream position only for consumption
lag on transactional Kafka topics (#20024)
0007e8b9e4a is described below
commit 0007e8b9e4a144eb8f102ca1c52f0af3af848963
Author: Xiang Fu <[email protected]>
AuthorDate: Sun Nov 8 10:07:48 2026 -0800
Use the empty-batch stream position only for consumption lag on
transactional Kafka topics (#20024)
A Kafka transaction's commit marker occupies an offset the client never
delivers, so on an idle transactional partition the consumer sits at the
high watermark while RealtimeSegmentDataManager's current offset stays one
behind it. The records lag in ConsumerPartitionState therefore never
reached 0, and the HIGHEST_STREAM_OFFSET_CONSUMED gauge stayed one behind.
RealtimeSegmentDataManager now records the next offset of a batch without
messages as the stream position when it is ahead of the current offset.
Only the partition lag state and the HIGHEST_STREAM_OFFSET_CONSUMED gauge
read it. The current offset, and with it every fetch, the catch-up, the
KEEP and ONLINE transitions and the committed end offset, stays one past
the last delivered record.
The kafka-4.0 KafkaPartitionLevelConsumer now reports the client position
on an empty poll under read_uncommitted too, not only read_committed. The
move is not validated, which is safe because no caller resumes from it:
the next poll with records still runs the data-loss check against the
caller's offset.
StatelessRealtimeSegmentWriter (re-ingestion) no longer moves to an empty
batch's next offset. It polls again from its own offset, so it cannot
adopt an unreported reset, and the loss check on the next batch with
records covers the gap.
The MessageBatch.getOffsetOfNextBatch Javadoc now says that an empty
batch's value can be ahead of the requested offset and is only used for
lag. No config, SPI signature, wire or segment metadata change.
---
.../realtime/RealtimeSegmentDataManager.java | 61 ++++++-
.../realtime/RealtimeSegmentDataManagerTest.java | 196 +++++++++++++++++++++
.../kafka40/KafkaPartitionLevelConsumer.java | 38 ++--
...fkaPartitionLevelConsumerDataLossCheckTest.java | 12 +-
...titionLevelConsumerDataLossIntegrationTest.java | 30 +++-
.../KafkaPartitionLevelConsumerSeekTest.java | 27 +++
.../writer/StatelessRealtimeSegmentWriter.java | 10 ++
.../writer/StatelessRealtimeSegmentWriterTest.java | 34 ++--
.../org/apache/pinot/spi/stream/MessageBatch.java | 5 +
9 files changed, 360 insertions(+), 53 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
index 29e4efff320..8641227c9cd 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
@@ -274,6 +274,10 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
private ServerIngestionOomProtectionManager
_serverIngestionOomProtectionManager;
private final MutableSegmentImpl _realtimeSegment;
private volatile StreamPartitionMsgOffset _currentOffset; // Next offset to
be consumed
+ // Furthest next offset reported by a batch without messages, e.g. one past
a Kafka transaction marker. Only feeds
+ // consumption lag and the highest-offset gauge (see getStreamPosition()).
Written by the consumer thread only.
+ @Nullable
+ private volatile StreamPartitionMsgOffset _emptyBatchStreamPosition;
private volatile State _state;
private volatile int _numRowsConsumed = 0;
private volatile int _numRowsIndexed = 0; // Can be different from
_numRowsConsumed when metrics update is enabled.
@@ -575,12 +579,7 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
if (_currentOffset.compareTo(lastUpdatedOffset) != 0) {
_idleTimer.markEventConsumed();
// We consumed something. Update the highest stream offset as well as
partition-consuming metric.
- // TODO Issue 5359 Need to find a way to bump metrics without getting
actual offset value.
- if (_currentOffset instanceof LongMsgOffset) {
- // TODO: only LongMsgOffset supplies long offset value.
- _serverMetrics.setValueOfTableGauge(_clientId,
ServerGauge.HIGHEST_STREAM_OFFSET_CONSUMED,
- ((LongMsgOffset) _currentOffset).getOffset());
- }
+ updateHighestStreamOffsetConsumedGauge();
lastUpdatedOffset =
_streamPartitionMsgOffsetFactory.create(_currentOffset);
} else if (endCriteriaReached) {
// At this point current offset has not moved because
processStreamEvents() has exited before processing a
@@ -602,6 +601,9 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
lastUpdatedOffset =
_streamPartitionMsgOffsetFactory.create(nextOffset);
} else {
// We did not consume any rows.
+ if (recordStreamPositionOfEmptyBatch(messageBatch)) {
+ updateHighestStreamOffsetConsumedGauge();
+ }
long timeSinceStreamLastCreatedOrConsumedMs =
_idleTimer.getTimeSinceStreamLastCreatedOrConsumedMs();
if (idleTimeoutMillis >= 0 && (timeSinceStreamLastCreatedOrConsumedMs
> idleTimeoutMillis)) {
@@ -628,6 +630,48 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
return true;
}
+ // TODO Issue 5359 Need to find a way to bump metrics without getting actual
offset value.
+ private void updateHighestStreamOffsetConsumedGauge() {
+ StreamPartitionMsgOffset streamPosition = getStreamPosition();
+ if (streamPosition instanceof LongMsgOffset) {
+ // TODO: only LongMsgOffset supplies long offset value.
+ _serverMetrics.setValueOfTableGauge(_clientId,
ServerGauge.HIGHEST_STREAM_OFFSET_CONSUMED,
+ ((LongMsgOffset) streamPosition).getOffset());
+ }
+ }
+
+ /// Records the next offset of a batch without messages when it is ahead of
[#getStreamPosition()], which happens
+ /// when the stream moved past offsets it never delivers, such as Kafka
transaction markers.
+ ///
+ /// @return `true` if the stream position moved
+ private boolean recordStreamPositionOfEmptyBatch(MessageBatch messageBatch) {
+ StreamPartitionMsgOffset nextBatchOffset =
messageBatch.getOffsetOfNextBatch();
+ if (nextBatchOffset == null ||
nextBatchOffset.compareTo(getStreamPosition()) <= 0) {
+ return false;
+ }
+ if (_segmentLogger.isDebugEnabled()) {
+ _segmentLogger.debug("Empty batch moved the stream position to {}
(current offset {})", nextBatchOffset,
+ _currentOffset);
+ }
+ _emptyBatchStreamPosition = nextBatchOffset;
+ return true;
+ }
+
+ /// Returns how far the stream has been read: `_currentOffset`, or the
furthest next offset reported by a batch
+ /// without messages when that is ahead of it. A Kafka consumer moves past
transaction markers without delivering
+ /// anything, so on a transactional topic a caught-up consumer sits one past
the last commit marker while
+ /// `_currentOffset` stays one past the last record. This position only
feeds consumption lag
+ /// ([#getConsumerPartitionState]) and
[ServerGauge#HIGHEST_STREAM_OFFSET_CONSUMED]. Fetching, catch-up and the end
+ /// offset a segment commits keep using `_currentOffset`, so an empty
batch's next offset never decides which records
+ /// a segment holds.
+ @VisibleForTesting
+ StreamPartitionMsgOffset getStreamPosition() {
+ StreamPartitionMsgOffset currentOffset = _currentOffset;
+ StreamPartitionMsgOffset emptyBatchStreamPosition =
_emptyBatchStreamPosition;
+ return emptyBatchStreamPosition != null &&
emptyBatchStreamPosition.compareTo(currentOffset) > 0
+ ? emptyBatchStreamPosition : currentOffset;
+ }
+
/// @param messageBatch batch of messages to process
/// @param idlePipeSleepTimeMillis wait time in case no messages were read
/// @return returns `true` if the process loop ended before processing the
batch, `false`
@@ -1176,12 +1220,13 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
return _lastConsumedTimestampMs;
}
- /// Returns the [ConsumerPartitionState] for the partition group.
+ /// Returns the [ConsumerPartitionState] for the partition group. Its
current offset is [#getStreamPosition()], so
+ /// the lag computed from it counts offsets the stream skipped without
delivering anything as consumed.
public Map<String, ConsumerPartitionState> getConsumerPartitionState(
@Nullable StreamPartitionMsgOffset latestMsgOffset) {
String partitionGroupId = String.valueOf(_partitionGroupId);
return Map.of(partitionGroupId,
- new ConsumerPartitionState(partitionGroupId, getCurrentOffset(),
getLastConsumedTimestamp(), latestMsgOffset,
+ new ConsumerPartitionState(partitionGroupId, getStreamPosition(),
getLastConsumedTimestamp(), latestMsgOffset,
_lastRowMetadata));
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
index ab360012d95..49daefd1ff2 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
@@ -25,6 +25,7 @@ import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -41,6 +42,7 @@ import
org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.io.FileUtils;
import org.apache.helix.HelixManager;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.metrics.ServerGauge;
import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
@@ -74,11 +76,15 @@ import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.metrics.PinotMetricUtils;
+import org.apache.pinot.spi.stream.BytesStreamMessage;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.LongMsgOffsetFactory;
import org.apache.pinot.spi.stream.MessageBatch;
+import org.apache.pinot.spi.stream.PartitionGroupConsumer;
import org.apache.pinot.spi.stream.PermanentConsumerException;
import org.apache.pinot.spi.stream.StreamConfigProperties;
+import org.apache.pinot.spi.stream.StreamMessage;
+import org.apache.pinot.spi.stream.StreamMessageMetadata;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
@@ -442,6 +448,185 @@ public class RealtimeSegmentDataManagerTest {
}
}
+ /// A batch without messages whose next offset is ahead of the current
offset means the stream moved past offsets it
+ /// never delivers (a Kafka consumer skipping a transaction commit marker,
for example). That position feeds lag and
+ /// the highest-offset gauge, while the current offset, which every fetch
starts from, stays where the last record
+ /// left it. A later batch that reports the requested offset, no offset, or
an offset behind does not move the
+ /// position back.
+ @Test
+ public void testEmptyBatchAheadOfCurrentOffsetMovesOnlyTheStreamPosition()
+ throws Exception {
+ ServerMetrics serverMetrics = mock(ServerMetrics.class);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(true, new TimeSupplier(), null, null, null,
serverMetrics)) {
+ segmentDataManager._stubConsumeLoop = false;
+ segmentDataManager._state.set(segmentDataManager,
RealtimeSegmentDataManager.State.INITIAL_CONSUMING);
+ // consumeLoop() is called directly, so set the time limit run() would
have set
+ segmentDataManager.setConsumeEndTime(System.currentTimeMillis() +
TimeUnit.HOURS.toMillis(1));
+ ScriptedPartitionGroupConsumer consumer = new
ScriptedPartitionGroupConsumer(segmentDataManager,
+ // The stream skipped START_OFFSET_VALUE and START_OFFSET_VALUE + 1
without delivering anything
+ emptyBatch(START_OFFSET_VALUE + 2),
+ // Nothing moved
+ emptyBatch(START_OFFSET_VALUE + 2),
+ // A stream that does not report where the next batch starts
+ emptyBatch(null),
+ // Behind the stream position
+ emptyBatch(START_OFFSET_VALUE + 1));
+ segmentDataManager.setPartitionGroupConsumer(consumer);
+
+ segmentDataManager.consumeLoop();
+
+ assertEquals(consumer._requestedOffsets,
+ List.of(START_OFFSET_VALUE, START_OFFSET_VALUE, START_OFFSET_VALUE,
START_OFFSET_VALUE));
+ assertEquals(((LongMsgOffset)
segmentDataManager.getCurrentOffset()).getOffset(), START_OFFSET_VALUE);
+ assertEquals(((LongMsgOffset)
segmentDataManager.getStreamPosition()).getOffset(), START_OFFSET_VALUE + 2);
+ assertEquals(((LongMsgOffset)
segmentDataManager.getConsumerPartitionState(null)
+
.get(String.valueOf(PARTITION_GROUP_ID)).getCurrentOffset()).getOffset(),
START_OFFSET_VALUE + 2);
+ verify(serverMetrics).setValueOfTableGauge(anyString(),
eq(ServerGauge.HIGHEST_STREAM_OFFSET_CONSUMED),
+ anyLong());
+ verify(serverMetrics).setValueOfTableGauge(anyString(),
eq(ServerGauge.HIGHEST_STREAM_OFFSET_CONSUMED),
+ eq(START_OFFSET_VALUE + 2));
+ assertEquals(segmentDataManager.getSegment().getNumDocsIndexed(), 0);
+ }
+ }
+
+ /// Records still move the current offset after the stream position moved
ahead of it, and the stream position
+ /// follows: a single-record transaction after an idle period on a
transactional topic, then its commit marker.
+ @Test
+ public void testRecordAfterEmptyBatchMovesTheCurrentOffset()
+ throws Exception {
+ ServerMetrics serverMetrics = mock(ServerMetrics.class);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(true, new TimeSupplier(), null, null, null,
serverMetrics)) {
+ segmentDataManager._stubConsumeLoop = false;
+ segmentDataManager._state.set(segmentDataManager,
RealtimeSegmentDataManager.State.INITIAL_CONSUMING);
+ segmentDataManager.setConsumeEndTime(System.currentTimeMillis() +
TimeUnit.HOURS.toMillis(1));
+ ScriptedPartitionGroupConsumer consumer = new
ScriptedPartitionGroupConsumer(segmentDataManager,
+ // Commit marker at START_OFFSET_VALUE
+ emptyBatch(START_OFFSET_VALUE + 1),
+ // The next transaction's record, then its marker at
START_OFFSET_VALUE + 2
+ new ScriptedMessageBatch(START_OFFSET_VALUE + 2, START_OFFSET_VALUE
+ 1),
+ emptyBatch(START_OFFSET_VALUE + 3));
+ segmentDataManager.setPartitionGroupConsumer(consumer);
+
+ segmentDataManager.consumeLoop();
+
+ assertEquals(consumer._requestedOffsets, List.of(START_OFFSET_VALUE,
START_OFFSET_VALUE, START_OFFSET_VALUE + 2));
+ assertEquals(((LongMsgOffset)
segmentDataManager.getCurrentOffset()).getOffset(), START_OFFSET_VALUE + 2);
+ assertEquals(((LongMsgOffset)
segmentDataManager.getStreamPosition()).getOffset(), START_OFFSET_VALUE + 3);
+ for (long offset = START_OFFSET_VALUE + 1; offset <= START_OFFSET_VALUE
+ 3; offset++) {
+ verify(serverMetrics).setValueOfTableGauge(anyString(),
eq(ServerGauge.HIGHEST_STREAM_OFFSET_CONSUMED),
+ eq(offset));
+ }
+ }
+ }
+
+ /// A replica whose stream position moved past a commit marker still posts,
and commits, the offset one past its
+ /// last record: the end offset of a segment never depends on an empty batch.
+ @Test
+ public void testSegmentEndsOnePastLastRecordAfterEmptyBatch()
+ throws Exception {
+ TimeSupplier timeSupplier = new TimeSupplier();
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
+ createFakeSegmentManager(true, timeSupplier, null, null, null)) {
+ segmentDataManager._stubConsumeLoop = false;
+ segmentDataManager._state.set(segmentDataManager,
RealtimeSegmentDataManager.State.INITIAL_CONSUMING);
+ segmentDataManager.setConsumeEndTime(timeSupplier.get() +
TimeUnit.HOURS.toMillis(1));
+ // A record, then an empty poll that skipped its commit marker, then the
time limit
+ ScriptedPartitionGroupConsumer consumer =
+ new ScriptedPartitionGroupConsumer(() ->
timeSupplier.add(TimeUnit.HOURS.toMillis(2)),
+ new ScriptedMessageBatch(START_OFFSET_VALUE + 1,
START_OFFSET_VALUE),
+ emptyBatch(START_OFFSET_VALUE + 2));
+ segmentDataManager.setPartitionGroupConsumer(consumer);
+ segmentDataManager._responses.add(new SegmentCompletionProtocol.Response(
+ new SegmentCompletionProtocol.Response.Params()
+ .withStreamPartitionMsgOffset(Long.toString(START_OFFSET_VALUE +
1))
+
.withStatus(SegmentCompletionProtocol.ControllerResponseStatus.COMMIT)));
+
+ segmentDataManager.createPartitionConsumer().run();
+
+ assertEquals(segmentDataManager._postedOffsets,
List.of(START_OFFSET_VALUE + 1));
+ assertTrue(segmentDataManager._commitSegmentCalled);
+ assertEquals(segmentDataManager._state.get(segmentDataManager),
RealtimeSegmentDataManager.State.COMMITTED);
+ assertEquals(((LongMsgOffset)
segmentDataManager.getCurrentOffset()).getOffset(), START_OFFSET_VALUE + 1);
+ assertEquals(((LongMsgOffset)
segmentDataManager.getStreamPosition()).getOffset(), START_OFFSET_VALUE + 2);
+ }
+ }
+
+ /// A batch without messages; `nextBatchOffset` is where the stream consumer
stands after the poll.
+ private static MessageBatch<byte[]> emptyBatch(@Nullable Long
nextBatchOffset) {
+ return new ScriptedMessageBatch(nextBatchOffset);
+ }
+
+ /// A batch of empty messages at the given offsets (each ending at the
following offset), reporting the given
+ /// next-batch offset (`null` for a stream that does not report one). A
consumed message fails to decode and is
+ /// dropped as an erroneous row, which still moves the current offset past
it.
+ private static class ScriptedMessageBatch implements MessageBatch<byte[]> {
+ private final Long _nextBatchOffset;
+ private final long[] _messageOffsets;
+
+ ScriptedMessageBatch(@Nullable Long nextBatchOffset, long...
messageOffsets) {
+ _nextBatchOffset = nextBatchOffset;
+ _messageOffsets = messageOffsets;
+ }
+
+ @Override
+ public int getMessageCount() {
+ return _messageOffsets.length;
+ }
+
+ @Override
+ public StreamMessage<byte[]> getStreamMessage(int index) {
+ long offset = _messageOffsets[index];
+ return new BytesStreamMessage(new byte[0], new
StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(offset), new LongMsgOffset(offset +
1)).build());
+ }
+
+ @Override
+ public StreamPartitionMsgOffset getOffsetOfNextBatch() {
+ return _nextBatchOffset != null ? new LongMsgOffset(_nextBatchOffset) :
null;
+ }
+
+ @Override
+ public long getSizeInBytes() {
+ return 0;
+ }
+ }
+
+ /// Hands out the given batches in order, recording the offset requested for
each, and runs `afterLastBatch` once the
+ /// last one has been handed out (by default it stops the data manager, so
consumeLoop() returns).
+ private static class ScriptedPartitionGroupConsumer implements
PartitionGroupConsumer {
+ final List<Long> _requestedOffsets = new ArrayList<>();
+ private final Runnable _afterLastBatch;
+ private final LinkedList<MessageBatch<byte[]>> _batches;
+
+ @SafeVarargs
+ ScriptedPartitionGroupConsumer(FakeRealtimeSegmentDataManager
segmentDataManager,
+ MessageBatch<byte[]>... batches) {
+ this(segmentDataManager::stop, batches);
+ }
+
+ @SafeVarargs
+ ScriptedPartitionGroupConsumer(Runnable afterLastBatch,
MessageBatch<byte[]>... batches) {
+ _afterLastBatch = afterLastBatch;
+ _batches = new LinkedList<>(List.of(batches));
+ }
+
+ @Override
+ public MessageBatch<byte[]> fetchMessages(StreamPartitionMsgOffset
startOffset, int timeoutMs) {
+ _requestedOffsets.add(((LongMsgOffset) startOffset).getOffset());
+ MessageBatch<byte[]> batch = _batches.remove();
+ if (_batches.isEmpty()) {
+ _afterLastBatch.run();
+ }
+ return batch;
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+
private FakeRealtimeSegmentDataManager createFakeSegmentManager(boolean
noUpsert, TimeSupplier timeSupplier,
@Nullable String maxRows, @Nullable String maxDuration, @Nullable
TableConfig tableConfig)
throws Exception {
@@ -2002,6 +2187,8 @@ public class RealtimeSegmentDataManagerTest {
private Field _streamMsgOffsetFactory;
public LinkedList<LongMsgOffset> _consumeOffsets = new LinkedList<>();
public LinkedList<SegmentCompletionProtocol.Response> _responses = new
LinkedList<>();
+ // Offsets posted with segmentConsumed(), which the real
postSegmentConsumedMsg() reads from getCurrentOffset()
+ public List<Long> _postedOffsets = new ArrayList<>();
public boolean _commitSegmentCalled = false;
public boolean _buildSegmentCalled = false;
public boolean _failSegmentBuild = false;
@@ -2069,6 +2256,14 @@ public class RealtimeSegmentDataManagerTest {
realtimeSegmentField.set(this, realtimeSegment);
}
+ /// Replaces the stream consumer so tests can script the batches
consumeLoop() sees.
+ public void setPartitionGroupConsumer(PartitionGroupConsumer
partitionGroupConsumer)
+ throws Exception {
+ Field partitionGroupConsumerField =
RealtimeSegmentDataManager.class.getDeclaredField("_partitionGroupConsumer");
+ partitionGroupConsumerField.setAccessible(true);
+ partitionGroupConsumerField.set(this, partitionGroupConsumer);
+ }
+
/// Replaces the upsert metadata manager, which the constructor always
leaves null.
public void
setPartitionUpsertMetadataManager(PartitionUpsertMetadataManager
partitionUpsertMetadataManager)
throws Exception {
@@ -2146,6 +2341,7 @@ public class RealtimeSegmentDataManagerTest {
@Override
protected SegmentCompletionProtocol.Response postSegmentConsumedMsg() {
+ _postedOffsets.add(((LongMsgOffset) getCurrentOffset()).getOffset());
return _responses.remove();
}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java
index a0fec35ebe4..c69ef090857 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java
@@ -57,10 +57,10 @@ public class KafkaPartitionLevelConsumer extends
KafkaPartitionLevelConnectionHa
// Offset the consumer is positioned to read NEXT. -1 means the consumer has
not been
// positioned for any caller-requested startOffset yet (we have not issued a
seek).
- // After a successful fetch, this is lastRecord.offset + 1. After an empty
fetch under
- // read_committed (where the batch may have been filtered as aborted), this
is the
+ // After a successful fetch, this is lastRecord.offset + 1. After an empty
fetch, this is the
// consumer's actual KafkaConsumer.position(), which may have advanced past
the caller's
- // startOffset even though zero records were returned.
+ // startOffset even though zero records were returned (transaction control
records under
+ // either isolation level, aborted records under read_committed).
private long _nextReadOffset = -1;
// The caller's startOffset on the previous fetchMessages() call. Used to
distinguish
// "caller is passing the same startOffset back because
RealtimeSegmentDataManager's
@@ -107,10 +107,10 @@ public class KafkaPartitionLevelConsumer extends
KafkaPartitionLevelConnectionHa
// the repeated-startOffset-after-empty-fetch case described below.
//
// We only suppress a re-seek when the caller passed the same startOffset
after the
- // previous successful fetch returned zero records. That is the
read_committed
- // empty-poll case: RealtimeSegmentDataManager._currentOffset did not
advance, but
- // KafkaConsumer.position() may have advanced through aborted records.
Seeking back to
- // startOffset would undo that progress and wedge consumption forever.
+ // previous successful fetch returned zero records. That is the empty-poll
case:
+ // RealtimeSegmentDataManager._currentOffset did not advance, but
KafkaConsumer.position()
+ // may have advanced through control or aborted records. Seeking back to
startOffset would
+ // undo that progress (and wedge consumption forever in an aborted region).
//
// If the previous poll returned records, a repeated startOffset means the
caller did
// not process them (for example, the segment time limit fired before
index 0). We must
@@ -151,18 +151,18 @@ public class KafkaPartitionLevelConsumer extends
KafkaPartitionLevelConnectionHa
}
lastMessageMetadata = messageMetadata;
}
- } else if (_isReadCommitted) {
- // No records returned. With read_committed isolation, the underlying
KafkaConsumer's
- // internal position may still have advanced past offsets filtered out
as aborted
- // transactional records, so snap _nextReadOffset to the consumer's
actual position.
- // RealtimeSegmentDataManager doesn't advance _currentOffset on empty
batches and will
- // call us again with the same startOffset; without this update the
seek-check would
- // re-seek to startOffset and undo the consumer's progress through the
aborted region.
- //
- // For read_uncommitted (the default), empty polls can never advance the
internal
- // position past records that didn't exist, so we skip the position()
call in the
- // hot path. The call is bounded by the same timeout as poll() so a sick
broker
- // can't stall this loop unbounded.
+ } else {
+ // No records returned. The underlying KafkaConsumer's internal position
may still have
+ // advanced past offsets it never delivers: transaction control records
(commit/abort
+ // markers) under either isolation level, and records filtered out as
aborted under
+ // read_committed. Snap _nextReadOffset to the consumer's actual
position and report it as
+ // the next batch's offset. RealtimeSegmentDataManager only uses it for
consumption lag
+ // and calls us again with the same startOffset, which the seek-check
above does not undo.
+ // The move is not validated (it may also be an out-of-range reset),
which is fine because
+ // no caller resumes from it: the next poll that returns records checks
its first offset
+ // against the caller's startOffset below. The call is bounded by the
same timeout as
+ // poll() so a sick broker can't stall this loop unbounded; after a poll
it is normally
+ // answered locally.
long currentPosition;
try {
currentPosition = _consumer.position(_topicPartition,
Duration.ofMillis(timeoutMs));
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossCheckTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossCheckTest.java
index 166a2c9838f..7b8129f1995 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossCheckTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossCheckTest.java
@@ -42,6 +42,7 @@ import org.testng.annotations.Test;
import static org.apache.kafka.common.record.RecordBatch.NO_TIMESTAMP;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyCollection;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -140,22 +141,23 @@ public class KafkaPartitionLevelConsumerDataLossCheckTest
{
}
/// An out-of-range reset can surface on an empty poll: the client jumps to
the new log start without returning a
- /// record, so that batch cannot be told from one that only skipped
transaction markers and reports no loss. Under
- /// read_uncommitted the empty batch reports the requested offset as the
next one, so the caller asks for it again
- /// and the loss is reported by the next poll that returns records: its
first offset is past the requested one and
- /// the log now starts above it.
+ /// record, so that batch cannot be told from one that only skipped
transaction markers and reports no loss. The
+ /// empty batch reports the client position as the next one, which callers
only use for consumption lag, so the
+ /// caller asks for its own offset again and the loss is reported by the
next poll that returns records: its first
+ /// offset is past the requested one and the log now starts above it.
@Test
public void testResetSeenOnEmptyPollIsReportedOnNextDataPoll()
throws Exception {
Consumer<Bytes, Bytes> mockConsumer = mock(Consumer.class);
when(mockConsumer.poll(any(Duration.class))).thenReturn(records(),
records(record(150L)));
+ when(mockConsumer.position(eq(TOPIC_PARTITION),
any(Duration.class))).thenReturn(150L);
when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class)))
.thenReturn(Map.of(TOPIC_PARTITION, 150L));
try (KafkaPartitionLevelConsumer consumer =
createConsumerWithMock(getStreamConfig(null), mockConsumer)) {
KafkaMessageBatch emptyBatch = consumer.fetchMessages(new
LongMsgOffset(100L), TIMEOUT_MS);
assertEquals(emptyBatch.getMessageCount(), 0);
- assertEquals(emptyBatch.getOffsetOfNextBatch().toString(), "100");
+ assertEquals(emptyBatch.getOffsetOfNextBatch().toString(), "150");
assertFalse(emptyBatch.hasDataLoss());
KafkaMessageBatch batch = consumer.fetchMessages(new
LongMsgOffset(100L), TIMEOUT_MS);
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossIntegrationTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossIntegrationTest.java
index ae1d703fece..525b7dec329 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossIntegrationTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossIntegrationTest.java
@@ -52,7 +52,8 @@ import static org.testng.Assert.assertTrue;
/// 2. Delete offsets (advance the log start via
[EmbeddedKafkaCluster#deleteRecordsBeforeOffset]) and
/// confirm data loss IS reported.
/// 3. Replay the production pattern: a caught-up consumer on a low-traffic
transactional topic with one
-/// record per transaction, which sees every new record one offset past
where it was parked.
+/// record per transaction, which sees every new record one offset past
where it was parked, and whose
+/// empty polls report the position past the commit marker (the high
watermark) for the consumption lag.
///
/// Speed/stability: setup uses only synchronous broker calls
(createTopics().all().get(),
/// commitTransaction()/flush(), deleteRecords().all().get()), so there are no
fixed sleeps. Reads
@@ -200,9 +201,9 @@ public class
KafkaPartitionLevelConsumerDataLossIntegrationTest {
assertEquals(firstOffset(batch), 0);
assertFalse(batch.hasDataLoss());
- // Caught up: the caller asks for offset 1, the marker. The client skips
it and returns nothing.
- KafkaMessageBatch emptyBatch = consumer.fetchMessages(new
LongMsgOffset(1), FETCH_TIMEOUT_MS);
- assertEquals(emptyBatch.getMessageCount(), 0);
+ // Caught up: the caller asks for offset 1, the marker. The client skips
it and returns nothing, and the empty
+ // batch reports the client position, 2, which is the high watermark:
the lag computed from it is 0.
+ KafkaMessageBatch emptyBatch = fetchUntilNextBatchOffset(consumer, 1, 2);
assertFalse(emptyBatch.hasDataLoss());
// Transaction 2: record at offset 2, marker at 3. The caller still asks
for offset 1, as
@@ -237,10 +238,31 @@ public class
KafkaPartitionLevelConsumerDataLossIntegrationTest {
return batch;
}
+ /// Polls repeatedly at the same startOffset until the (empty) batch reports
`expectedNextBatchOffset` as the
+ /// start of the next batch, i.e. until the consumer has observed and
skipped the control record(s) at/after
+ /// startOffset, or the wait budget elapses. Fails if a record shows up
instead.
+ private KafkaMessageBatch
fetchUntilNextBatchOffset(KafkaPartitionLevelConsumer consumer, long
startOffset,
+ long expectedNextBatchOffset) {
+ long deadlineMs = System.currentTimeMillis() + FETCH_MAX_WAIT_MS;
+ KafkaMessageBatch batch = consumer.fetchMessages(new
LongMsgOffset(startOffset), FETCH_TIMEOUT_MS);
+ while (batch.getMessageCount() == 0 && nextBatchOffset(batch) <
expectedNextBatchOffset
+ && System.currentTimeMillis() < deadlineMs) {
+ batch = consumer.fetchMessages(new LongMsgOffset(startOffset),
FETCH_TIMEOUT_MS);
+ }
+ assertEquals(batch.getMessageCount(), 0, "Expected no record at/after
offset " + startOffset + " yet");
+ assertEquals(nextBatchOffset(batch), expectedNextBatchOffset,
+ "Expected the empty batch to report the consumer position past the
skipped control record");
+ return batch;
+ }
+
private static long firstOffset(KafkaMessageBatch batch) {
return Long.parseLong(batch.getFirstMessageOffset().toString());
}
+ private static long nextBatchOffset(KafkaMessageBatch batch) {
+ return Long.parseLong(batch.getOffsetOfNextBatch().toString());
+ }
+
private StreamConfig streamConfig(String topic, String isolationLevel,
String autoOffsetReset) {
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put("streamType", "kafka");
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerSeekTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerSeekTest.java
index 26d946b3ea8..86f8bedd664 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerSeekTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerSeekTest.java
@@ -39,12 +39,14 @@ import org.testng.annotations.Test;
import static org.apache.kafka.common.record.RecordBatch.NO_TIMESTAMP;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyCollection;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
/// Regression tests for [KafkaPartitionLevelConsumer] seek decisions.
@@ -87,6 +89,31 @@ public class KafkaPartitionLevelConsumerSeekTest {
verify(mockConsumer, times(1)).seek(TOPIC_PARTITION, 0L);
}
+ /// Under read_uncommitted an empty poll can still move the consumer past a
transaction control record. The empty
+ /// batch must report that position as the start of the next batch, and the
repeated startOffset must not undo
+ /// the progress with a re-seek.
+ @Test
+ public void
testRepeatedStartOffsetAfterReadUncommittedEmptyFetchDoesNotSeekAgain() {
+ Consumer<Bytes, Bytes> mockConsumer = mock(Consumer.class);
+ ConsumerRecords<Bytes, Bytes> emptyRecords =
+ new ConsumerRecords<>(Map.of(TOPIC_PARTITION,
List.<ConsumerRecord<Bytes, Bytes>>of()), Map.of());
+ // The client skipped the commit marker at offset 1 without returning
anything, so its position is now 2
+ when(mockConsumer.poll(any(Duration.class))).thenReturn(emptyRecords,
records(record(2L)));
+ when(mockConsumer.position(eq(TOPIC_PARTITION),
any(Duration.class))).thenReturn(2L);
+ when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class)))
+ .thenReturn(Map.of(TOPIC_PARTITION, 0L));
+
+ KafkaPartitionLevelConsumer consumer =
createConsumerWithMock(getStreamConfig(null), mockConsumer);
+ KafkaMessageBatch emptyBatch = consumer.fetchMessages(new
LongMsgOffset(1L), 10000);
+ KafkaMessageBatch nextBatch = consumer.fetchMessages(new
LongMsgOffset(1L), 10000);
+
+ assertEquals(emptyBatch.getMessageCount(), 0);
+ assertEquals(emptyBatch.getOffsetOfNextBatch().toString(), "2");
+ assertEquals(nextBatch.getFirstMessageOffset().toString(), "2");
+ assertFalse(nextBatch.hasDataLoss());
+ verify(mockConsumer, times(1)).seek(TOPIC_PARTITION, 1L);
+ }
+
private static KafkaPartitionLevelConsumer
createConsumerWithMock(StreamConfig streamConfig,
Consumer<Bytes, Bytes> mockConsumer) {
class FakeKafkaPartitionLevelConsumer extends KafkaPartitionLevelConsumer {
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java
index 9355236c6b2..67a098c515c 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java
@@ -84,6 +84,12 @@ import org.slf4j.LoggerFactory;
/// otherwise silently miss part of its offset range. The segment then stays
in ERROR until the operator restores the
/// range from another source or deletes the segment to accept the loss.
///
+/// A batch without unfiltered messages never moves the writer, whatever next
offset it reports: the writer polls again
+/// from its own offset. A Kafka consumer reports its client position there,
which can lie past transaction markers,
+/// but also past deleted records after an out-of-range reset it has not
reported as loss yet. The loss check of the
+/// next batch with records then covers the whole gap. Reaching the end offset
never needs that position, since the
+/// data manager only moves its own offset on fetched records.
+///
/// TODO:
/// 1. Clean up this class and only keep the necessary parts.
/// 2. Use a different segment impl for better performance because it
doesn't need to serve queries.
@@ -281,6 +287,10 @@ public class StatelessRealtimeSegmentWriter implements
Closeable {
+ "the stream no longer holds this range (e.g. it aged out of
retention), so the segment cannot be "
+ "rebuilt from it. Restore the range from another source, or
delete the segment to accept the loss",
_currentOffset, _endOffset);
+ if (messageBatch.getUnfilteredMessageCount() == 0) {
+ // Poll again from the same offset (see the class Javadoc)
+ continue;
+ }
int messageCount = messageBatch.getMessageCount();
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriterTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriterTest.java
index 5ae086e4d19..50fc8c136e0 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriterTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriterTest.java
@@ -115,28 +115,27 @@ public class StatelessRealtimeSegmentWriterTest {
ServerMetrics.register(mock(ServerMetrics.class));
}
- /// A batch without unfiltered messages moves the writer to the next offset
it reports: a consumer can move past
- /// offsets it never delivers (Kafka transaction markers, aborted records
under read_committed), which may land
- /// below, at or past the end offset. A batch that reports the requested
offset leaves it alone. Every such replay
- /// completes with exactly the records below the end, and one whose tail
holds only skipped offsets completes as soon
- /// as the reported position reaches the end (the scripted streams below end
there).
+ /// A batch without unfiltered messages never moves the writer, whatever
next offset it reports: a Kafka consumer
+ /// reports its position there, past skipped transaction markers or aborted
records, and the writer polls again from
+ /// its own offset. Fetched records, filtered ones included, move it as
usual. Every such replay completes with
+ /// exactly the records below the end, including one whose empty batches
report a position at or past the end: the
+ /// next batch with records completes it.
@DataProvider
public Object[][] validReplays() {
return new Object[][]{
{"records", List.of(records(0, 120)), 120, List.of(0L)},
{"nonmoving", List.of(records(0, 100), empty(100, 0, false),
records(100, 120)), 120, List.of(0L, 100L, 100L)},
- // Commit marker at 100: the empty poll moves the consumer to 101, the
next one delivers 101..119
- {"marker", List.of(records(0, 100), empty(101, 0, false), records(101,
120)), 119, List.of(0L, 100L, 101L)},
+ // Commit marker at 100: the empty poll moves the consumer to 101, the
writer asks for 100 again and gets
+ // 101..119
+ {"marker", List.of(records(0, 100), empty(101, 0, false), records(101,
120)), 119, List.of(0L, 100L, 100L)},
{"markerThenFiltered", List.of(records(0, 100), empty(101, 0, false),
empty(106, 5, false), records(106, 120)),
- 114, List.of(0L, 100L, 101L, 106L)},
+ 114, List.of(0L, 100L, 100L, 106L)},
// Five fetched tombstones/filtered records legitimately advance the
position without decoded rows
{"filtered", List.of(records(0, 100), empty(105, 5, false),
records(105, 120)), 115, List.of(0L, 100L, 105L)},
- // Markers fill the tail of the range: the empty advance stays below
the end (the next record lands at the
- // end), or reaches or crosses the end, which completes the replay
with no further record
- {"markerTailBelowEnd", List.of(records(0, 100), empty(110, 0, false),
records(120, 121)), 100,
- List.of(0L, 100L, 110L)},
- {"markerTailAtEnd", List.of(records(0, 100), empty(120, 0, false)),
100, List.of(0L, 100L)},
- {"markerTailPastEnd", List.of(records(0, 100), empty(150, 0, false)),
100, List.of(0L, 100L)},
+ // Markers fill the tail of the range: the empty batches report
positions below, at and past the end, and the
+ // record at the end completes the replay
+ {"markerTail", List.of(records(0, 100), empty(110, 0, false),
empty(120, 0, false), empty(150, 0, false),
+ records(120, 121)), 100, List.of(0L, 100L, 100L, 100L, 100L)},
{"alreadyAtEnd", List.of(), 0, List.of()}
};
}
@@ -185,7 +184,8 @@ public class StatelessRealtimeSegmentWriterTest {
}
/// A non-empty batch that reports data loss is not indexed, including the
case where an empty poll moved the
- /// consumer and the writer first (the loss check of the data poll covers
the gap from the adopted offset).
+ /// consumer first (the writer stays at its own offset, so the loss check of
the data poll covers the gap from
+ /// there).
@DataProvider
public Object[][] lossAfterMarker() {
return new Object[][]{{false}, {true}};
@@ -206,8 +206,8 @@ public class StatelessRealtimeSegmentWriterTest {
assertFalse(replay.writer().isSuccess());
assertEquals(indexedRows(replay.writer()), 100, "The loss batch must not
be indexed");
assertTrue(replay.writer().getConsumptionException().getMessage().contains("data
loss"));
- assertEquals(((LongMsgOffset) field(replay.writer(),
"_currentOffset")).getOffset(), afterMarker ? 101L : 100L);
- assertEquals(replay.requestedOffsets(), afterMarker ? List.of(0L, 100L,
101L) : List.of(0L, 100L));
+ assertEquals(((LongMsgOffset) field(replay.writer(),
"_currentOffset")).getOffset(), 100L);
+ assertEquals(replay.requestedOffsets(), afterMarker ? List.of(0L, 100L,
100L) : List.of(0L, 100L));
verify(replay.consumer(), times(1)).close();
}
}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java
index 55c485fbfec..920b07010be 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java
@@ -42,6 +42,11 @@ public interface MessageBatch<T> {
StreamMessage<T> getStreamMessage(int index);
/// Returns the start offset of the next batch.
+ ///
+ /// For a batch without messages this can be ahead of the requested start
offset when the stream moved past offsets
+ /// it never delivers (e.g. Kafka transaction control records). Pinot only
uses that value to report how far the
+ /// stream has been read (consumption lag): the next fetch still starts from
the requested offset, so a value past
+ /// undelivered messages cannot skip them.
StreamPartitionMsgOffset getOffsetOfNextBatch();
/// Returns the offset of the first message (including tombstone) in the
batch.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
