xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4369051914
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -605,6 +616,65 @@ 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 consumer position for INITIAL
consumption or ONLINE catch-up.
+ /// Transaction markers and aborted records can move a stream's position
without delivering rows, so adoption
+ /// keeps position and lag current. But a position, including a batch with
`hasDataLoss() == false`, does not prove
+ /// the skipped interval contained no rows: validation may be unavailable,
or the topic may have been compacted.
+ /// Remember that uncertainty for local-reuse CRC validation even when the
new position is below the final offset.
+ ///
+ /// Protocol CATCHING_UP can subsequently become the committer, so it must
not synthesize progress from an empty
+ /// batch. ONLINE catch-up can cap the position at the committed final
offset, because any local reuse is guarded
+ /// by the committed CRC. A reported loss is not capped; ordinary overshoot
recovery still applies.
+ ///
+ /// @return `true` if `_currentOffset` was advanced
+ private boolean adoptNextBatchOffsetOfEmptyBatch(MessageBatch messageBatch) {
+ if (_state == State.CATCHING_UP) {
+ return false;
+ }
+ StreamPartitionMsgOffset nextBatchOffset =
messageBatch.getOffsetOfNextBatch();
+ if (nextBatchOffset == null || nextBatchOffset.compareTo(_currentOffset)
<= 0) {
+ return false;
+ }
+ if (_state == State.CONSUMING_TO_ONLINE && _finalOffset != null &&
!messageBatch.hasDataLoss()
+ && nextBatchOffset.compareTo(_finalOffset) > 0) {
+ nextBatchOffset = _finalOffset;
+ }
+ if (_segmentLogger.isDebugEnabled()) {
+ _segmentLogger.debug("Empty batch moved the stream position. Advancing
from {} to {}", _currentOffset,
+ nextBatchOffset);
+ }
+ _hasUnverifiedOffsetAdvance = true;
+ _currentOffset = nextBatchOffset;
+ return true;
+ }
+
+ /// ONLINE catch-up can stop before a message ending past the committed
final offset. This only establishes
+ /// offset equality, not row equality, so record the unverified advance
before publishing the final offset.
+ /// Protocol CATCHING_UP must retain ordinary record advancement and
overshoot handling instead of manufacturing
+ /// a target match that could make this replica the committer. Reported loss
also retains overshoot recovery.
+ ///
+ /// @return `true` if consumption must stop before the message
+ private boolean reachedFinalOffsetBeforeMessage(MessageBatch messageBatch,
StreamPartitionMsgOffset nextOffset) {
+ if (_state != State.CONSUMING_TO_ONLINE || _finalOffset == null ||
messageBatch.hasDataLoss()
+ || _currentOffset.compareTo(_finalOffset) >= 0 ||
nextOffset.compareTo(_finalOffset) <= 0) {
Review Comment:
Nit: when a real message straddles _finalOffset (offset < final <
nextOffset), this settles at _finalOffset without indexing that message, which
is only safe because _hasUnverifiedOffsetAdvance then forces the CRC-guarded
download. That coupling is correct but subtle; a one-line note here pointing at
the _hasUnverifiedOffsetAdvance guard would help the next reader.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -605,6 +616,65 @@ 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 consumer position for INITIAL
consumption or ONLINE catch-up.
+ /// Transaction markers and aborted records can move a stream's position
without delivering rows, so adoption
+ /// keeps position and lag current. But a position, including a batch with
`hasDataLoss() == false`, does not prove
+ /// the skipped interval contained no rows: validation may be unavailable,
or the topic may have been compacted.
+ /// Remember that uncertainty for local-reuse CRC validation even when the
new position is below the final offset.
+ ///
+ /// Protocol CATCHING_UP can subsequently become the committer, so it must
not synthesize progress from an empty
+ /// batch. ONLINE catch-up can cap the position at the committed final
offset, because any local reuse is guarded
+ /// by the committed CRC. A reported loss is not capped; ordinary overshoot
recovery still applies.
+ ///
+ /// @return `true` if `_currentOffset` was advanced
+ private boolean adoptNextBatchOffsetOfEmptyBatch(MessageBatch messageBatch) {
+ if (_state == State.CATCHING_UP) {
Review Comment:
Excluding CATCHING_UP from empty-batch adoption is the right safety call.
One liveness edge worth a confirming thought: if a committer committed at a
marker-only offset, a CATCHING_UP follower can't reach _finalOffset by
consuming records and will poll empty batches (100ms sleep each) until the
Helix ONLINE transition arrives, which then recovers via the
CONSUMING_TO_ONLINE adoption + CRC-guarded download. Self-recovering and
low-CPU, so fine — just flagging in case you want a comment noting the ONLINE
path is the intended escape.
--
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]
