xiangfu0 commented on code in PR #20137:
URL: https://github.com/apache/pinot/pull/20137#discussion_r4451854840
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfig.java:
##########
@@ -46,6 +46,12 @@ public class StreamConfig {
public static final long DEFAULT_STREAM_CONNECTION_TIMEOUT_MILLIS = 30_000;
public static final int DEFAULT_STREAM_FETCH_TIMEOUT_MILLIS = 5_000;
public static final int DEFAULT_IDLE_TIMEOUT_MILLIS = 3 * 60 * 1000;
+ /// First backoff before a message whose decoder threw
[RetriableDecodeException] is fetched and decoded again. It
+ /// doubles on every retry, up to [#getDecodeRetryMaxBackoffMillis()].
+ public static final long INITIAL_DECODE_RETRY_BACKOFF_MILLIS = 100;
+ public static final long DEFAULT_DECODE_RETRY_MAX_BACKOFF_MILLIS = 10_000;
+ /// Retries a message whose decoder throws [RetriableDecodeException] until
it decodes.
+ public static final long DEFAULT_DECODE_RETRY_MAX_DURATION_MILLIS =
Long.MAX_VALUE;
Review Comment:
(MINOR, design question for owners) This default changes existing Confluent
tables from "drop the message" to "stall the partition indefinitely" when the
registry keeps returning 5xx or can't be reached. A typo'd registry host is one
case, since `UnknownHostException` is retriable. If the stall outlasts topic
retention, it loses far more data than the one message it protects. Consider a
bounded default (e.g. hours), or making the new behaviour opt-in per C1.6.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -663,6 +702,25 @@ private boolean processStreamEvents(MessageBatch
messageBatch, long idlePipeSlee
StreamPartitionMsgOffset offset = metadata.getOffset();
StreamPartitionMsgOffset nextOffset = metadata.getNextOffset();
int rowSizeInBytes = metadata.getRecordSerializedSize();
+ boolean retriable = decodedRow.isRetriable();
+ if (retriable && shouldRetryDecode()) {
+ // The message may decode later, e.g. once its schema registry is
reachable again. Stop here without moving
+ // _currentOffset past it, so that consumeLoop() backs off and fetches
it again instead of dropping it.
+ recordDecodeRetry(offset, decodedRow.getException());
Review Comment:
(MINOR) Each retry re-fetches a full batch. Before decoding,
`processStreamEvents` charges every message in it to `_partitionRateLimiter`
and the server-wide `_serverRateLimiter`, and `consumeLoop` adds them all to
`REALTIME_ROWS_FETCHED`. A stuck partition therefore uses shared server quota
and inflates the fetched metric without consuming anything. Consider skipping
the throttle and metric for rows a retried batch never gets to.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -776,7 +840,68 @@ private boolean processStreamEvents(MessageBatch
messageBatch, long idlePipeSlee
// If there were no messages to be fetched from stream, wait for a
little bit as to avoid hammering the stream
Uninterruptibles.sleepUninterruptibly(idlePipeSleepTimeMillis,
TimeUnit.MILLISECONDS);
}
- return prematureExit;
+ if (decodeRetryNeeded) {
+ return BatchOutcome.DECODE_RETRY_NEEDED;
+ }
+ return prematureExit ? BatchOutcome.STOPPED_EARLY : BatchOutcome.PROCESSED;
+ }
+
+ /// Returns whether the message whose decoder threw
`RetriableDecodeException` should be fetched and decoded again,
+ /// i.e. whether it has been retried for less than
[StreamConfig#getDecodeRetryMaxDurationMillis()]. Otherwise, it is
+ /// handled like any other decode failure.
+ private boolean shouldRetryDecode() {
+ long maxDurationMillis = _streamConfig.getDecodeRetryMaxDurationMillis();
+ if (_numDecodeRetries == 0) {
+ return maxDurationMillis > 0;
+ }
+ return now() - _decodeRetryStartTimeMs < maxDurationMillis;
+ }
+
+ /// Counts a retriable decode failure. The first one of a message is logged
at WARN with its cause and recorded as a
+ /// segment error; while the message keeps failing, its failures are logged
at WARN at most once a minute.
+ private void recordDecodeRetry(StreamPartitionMsgOffset offset, Exception
cause) {
+ _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.REALTIME_DECODE_RETRIES, 1L);
+ long nowMs = now();
+ if (_numDecodeRetries++ == 0) {
+ _decodeRetryStartTimeMs = nowMs;
+ _decodeRetryLastWarnTimeMs = nowMs;
+ _decodeRetryBackoffMs =
Math.min(StreamConfig.INITIAL_DECODE_RETRY_BACKOFF_MILLIS,
+ _streamConfig.getDecodeRetryMaxBackoffMillis());
+ String message = "Retriable decode error at offset: " + offset + ".
Fetching and decoding it again";
+ _segmentLogger.warn(message, cause);
+ _realtimeTableDataManager.addSegmentError(_segmentNameStr, new
SegmentErrorInfo(nowMs, message, cause));
+ } else if (nowMs - _decodeRetryLastWarnTimeMs >=
DECODE_RETRY_WARN_INTERVAL_MS) {
+ _decodeRetryLastWarnTimeMs = nowMs;
+ _segmentLogger.warn("Still retrying the message at offset: {} after {}
retriable decode failures in {}ms",
+ offset, _numDecodeRetries, nowMs - _decodeRetryStartTimeMs, cause);
+ } else if (_segmentLogger.isDebugEnabled()) {
+ _segmentLogger.debug("Retriable decode error at offset: {} (failure
{})", offset, _numDecodeRetries, cause);
+ }
+ }
+
+ /// Sleeps before a message whose decoder threw `RetriableDecodeException`
is fetched again, doubling the next backoff
+ /// up to the configured cap. The sleep ends at the consumption deadline of
states that have one, so that
+ /// [#endCriteriaReached()] can end the segment (or give up catching up) at
`_currentOffset`, which still points at
+ /// that message. [#stop()] interrupts the sleep.
+ private void backOffBeforeDecodeRetry()
+ throws InterruptedException {
+ if (_shouldStop) {
+ return;
+ }
+ long backoffMs = _decodeRetryBackoffMs;
+ _decodeRetryBackoffMs = Math.min(backoffMs * 2,
_streamConfig.getDecodeRetryMaxBackoffMillis());
+ if (_state == State.INITIAL_CONSUMING || _state ==
State.CONSUMING_TO_ONLINE) {
Review Comment:
(MINOR, test gap) The new tests only exercise retries in
`INITIAL_CONSUMING`. Nothing covers the cap at `_consumeEndTime` in
`CONSUMING_TO_ONLINE` (`catchupToFinalOffset` on the Helix thread), or
`CATCHING_UP`. `CATCHING_UP` has no deadline, so a stuck replica waits for the
ONLINE transition. One test per state would back the PR's "catch-up stays
bounded" claim.
--
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]
