xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4505412643
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -606,6 +631,82 @@ protected boolean consumeLoop()
return true;
}
+ // TODO Issue 5359 Need to find a way to bump metrics without getting actual
offset value.
+ private void updateHighestStreamOffsetConsumedGauge() {
+ if (_currentOffset instanceof LongMsgOffset) {
+ // TODO: only LongMsgOffset supplies long offset value.
+ _serverMetrics.setValueOfTableGauge(_clientId,
ServerGauge.HIGHEST_STREAM_OFFSET_CONSUMED,
+ ((LongMsgOffset) _currentOffset).getOffset());
+ }
+ }
+
+ /// Adopts an empty batch's advanced stream position. Transaction markers
and aborted records move a stream's
+ /// position without delivering rows, so adoption keeps the position and lag
current. The skipped offsets are treated
+ /// like a gap between two delivered messages, which needs no further proof
either: the stream reports a loss
+ /// through `hasDataLoss()`.
+ ///
+ /// @return `true` if `_currentOffset` was advanced
+ private boolean adoptNextBatchOffsetOfEmptyBatch(MessageBatch messageBatch) {
+ StreamPartitionMsgOffset nextBatchOffset =
messageBatch.getOffsetOfNextBatch();
+ if (nextBatchOffset == null || nextBatchOffset.compareTo(_currentOffset)
<= 0) {
+ return false;
+ }
+ nextBatchOffset = capAtFinalOffset(messageBatch, nextBatchOffset);
+ if (_segmentLogger.isDebugEnabled()) {
+ _segmentLogger.debug("Empty batch moved the stream position. Advancing
from {} to {}", _currentOffset,
+ nextBatchOffset);
+ }
+ _currentOffset = nextBatchOffset;
+ return true;
+ }
+
+ /// While catching up to a final offset, caps at it the position of a batch
that indexes no row (no messages, or
+ /// only filtered ones such as tombstones) when that position lies past the
final offset. The final offset then lies
+ /// inside an interval no replica indexes rows from, so this replica holds
exactly the rows below it. A committer can
+ /// stop at such an offset, e.g. one past a Kafka commit marker it skipped
on an empty poll. A batch reporting loss is
+ /// not capped and keeps the existing overshoot handling. A capped segment
is verified against the committed CRC on
+ /// KEEP.
+ @Nullable
+ private StreamPartitionMsgOffset capAtFinalOffset(MessageBatch messageBatch,
Review Comment:
Done, and the PR got a lot smaller. `_currentOffset`, and with it every
fetch and the committed end offset, stays at last-record+1. The data manager
now records the furthest next offset of an empty batch separately, and only the
lag in `getConsumerPartitionState` and the `HIGHEST_STREAM_OFFSET_CONSUMED`
gauge read it. The catch-up settling, the CRC check on KEEP, the
rolling-upgrade note and the consumer's validation of empty-poll moves are all
gone.
Nothing needs marker+1 for correctness. Replaying or resuming a segment that
ends at a marker fetches from the marker offset, and Kafka skips it. The one
place marker+1 helped is a narrow case that master already has: if retention
deletes the Kafka log segment holding the marker a segment ended on before
consumption resumes there, the controller and a read_uncommitted consumer
report that single offset as lost. I left that as it is and noted it in the
description.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java:
##########
@@ -42,6 +42,10 @@ default int getUnfilteredMessageCount() {
StreamMessage<T> getStreamMessage(int index);
/// Returns the start offset of the next batch.
+ ///
+ /// For a batch without messages this is the requested start offset, unless
the stream has moved past offsets it
+ /// will never deliver (e.g. Kafka transaction control records): it is then
the offset consumption resumes from,
+ /// and the consumer adopts it. It must never point past a message that has
not been delivered.
StreamPartitionMsgOffset getOffsetOfNextBatch();
Review Comment:
With the rework, neither the data manager nor the re-ingestion writer moves
its offset on an empty batch, so whatever a plugin returns there can't make
Pinot skip data. It only feeds the lag. The writer did follow it on master, for
the read_committed position Kafka reported. It now polls again from its own
offset, because the Kafka consumer reports its client position on every empty
poll and that move isn't validated. The `MessageBatch` Javadoc now says this,
so no opt-in or release note is needed.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -947,15 +1052,24 @@ public void run() {
"CompletionMode is DOWNLOAD but download URL is missing
for segment: {} and peer download "
+ "is disabled. Falling back to RETAINING mode",
_segmentNameStr);
}
+ // A final offset settled on is only as good as the stream's
loss reporting, which some streams (e.g.
+ // Pulsar, Kafka read_committed) never do, so a replica that
skipped the committer's last messages would
+ // settle too. Verify such a segment against the committed CRC.
+ SegmentZKMetadata committedSegmentZKMetadata =
+ _settledOnFinalOffset ?
_realtimeTableDataManager.fetchZKMetadata(_segmentNameStr) : null;
Review Comment:
This case is gone because there's no settling any more. The end offset is
always one past a delivered record, so catch-up, KEEP and the ONLINE transition
are back to master's behaviour.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
