xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4438826936
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java:
##########
@@ -272,6 +283,16 @@ public void run() {
// Fetch messages
MessageBatch messageBatch = _consumer.fetchMessages(_currentOffset,
_fetchTimeoutMs);
+ Preconditions.checkState(!messageBatch.hasDataLoss(), "Stream data
loss at offset %s while replaying to %s",
Review Comment:
Non-blocking: this makes re-ingestion fail closed once retention has passed
the start offset; before, it produced a segment with gaps. That's the safer
default, but a pauseless segment in that state now needs an operator to step
in. Worth confirming this is intended and noting it in the PR description.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -605,6 +620,74 @@ 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.
+ ///
+ /// A follower in protocol CATCHING_UP to a final offset the stream only
reaches by skipping (a committer that
+ /// adopted the position past a commit marker) therefore cannot reach it by
consuming. It keeps polling without
+ /// advancing, one idle sleep per empty batch, until the ONLINE transition
that follows the commit. That transition
+ /// is the intended way out: its CONSUMING_TO_ONLINE catch-up adopts the
position, and the local segment is reused
+ /// only on a matching committed CRC, otherwise downloaded.
+ ///
+ /// @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()
Review Comment:
Non-blocking, separate from the idle-stall thread above. On an *active*
topic, a CATCHING_UP follower at M with a final offset of M+1 (one past marker
M) can get the marker and record M+1 in one fetch. It indexes the record and
throws "Past max offset". In `run()` that posts `stoppedConsuming`, so the
controller marks the replica OFFLINE until the commit, and the replica
downloads. Recovery works, but end offsets could not trigger this on base.
Settling at `_finalOffset` in CATCHING_UP when the next record's *start* offset
is >= final and the batch reports no loss seems safe, since the rows below
final are the same. Otherwise this is worth a release note.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1412,12 +1533,21 @@ protected boolean buildSegmentAndReplace()
/// ordering); swapping in a divergent copy corrupts upsert metadata and
leaves replicas inconsistent otherwise. So
/// when `committedSegmentZKMetadata` carries a CRC, a mismatch discards the
local build (returns `false`) and the
/// caller downloads the committed segment. Skipped for pauseless tables and
segments whose CRC is not yet set (the
- /// COMMITTING window), so we never download a not-yet-uploaded segment (see
PR #17885).
+ /// COMMITTING window), so we never download a not-yet-uploaded segment (see
PR #17885). After an unverified offset
+ /// advance, those exceptions cannot establish row equality: local reuse
requires an enabled, matching committed
+ /// CRC check. Otherwise the caller must use the existing download/recovery
path.
///
- /// @param committedSegmentZKMetadata committed ZK metadata carrying the
CRC, or `null` to skip the check
+ /// @param committedSegmentZKMetadata committed ZK metadata carrying the
CRC, or `null` for legacy unflagged reuse
/// @return `true` if built and replaced locally; `false` if the build
failed or was rejected on a CRC mismatch
protected boolean buildSegmentAndReplace(@Nullable SegmentZKMetadata
committedSegmentZKMetadata)
throws Exception {
+ boolean crcCheckEnabled = committedSegmentZKMetadata != null &&
committedSegmentZKMetadata.getCrc() >= 0
+ && !PauselessConsumptionUtils.isPauselessEnabled(_tableConfig);
+ if (_hasUnverifiedOffsetAdvance && !crcCheckEnabled) {
Review Comment:
Non-blocking: KEEP calls `buildSegmentAndReplace()` with `null`, so once
this flag is set the replica goes to ERROR ("Could not build segment") and
downloads. Low-traffic transactional topics set the flag on most segments. The
controller only sends KEEP after the winner commits, so fetching ZK metadata
there (as the DOWNLOAD-mode branch does) would keep CRC-verified local reuse
instead of downloading.
--
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]
