raghavyadav01 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4504482393
##########
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:
This changes SPI behavior for third-party stream plugins: an empty batch's
next offset used to be ignored and is now adopted. Should this be opt-in per
plugin (e.g. a `MessageBatch` default method), or at least go in the release
note, so a custom plugin that returns a non-start offset on an empty poll
doesn't skip data?
##########
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:
For pauseless tables, or when the CRC isn't set yet,
`buildSegmentAndReplace` skips the CRC check, so a settled segment is kept
unverified. Since that's the Pulsar / read_committed silent-skip case, should a
settled segment download instead of building locally there?
##########
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:
Could we keep the committed end offset at last-record+1 and use the adopted
stream position only for lag and the offset gauge? That seems to avoid the
catch-up settling, the CRC check on KEEP and the rolling-upgrade cost. Or is
there a case that needs marker+1 as the end offset?
--
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]
