anshul98ks123 commented on code in PR #20137:
URL: https://github.com/apache/pinot/pull/20137#discussion_r4452036475
##########
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:
Keeping the default unlimited (owner's call). With the registry down or
mistyped every message fails, so the old behaviour dropped them all, while
holding loses nothing within retention and shows in ingestion delay and
`realtimeDecodeRetries`; `0` restores dropping. When a limit is set it now
bounds the whole outage, not each message (c33c9e0361).
##########
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:
Fixed in 5708c47af1: a batch fetched again for a retry is charged to both
rate limiters and `REALTIME_ROWS_FETCHED` only for the messages it processes,
after processing; other batches are unchanged.
##########
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:
Fixed in 422df32902: one test per state. In `CONSUMING_TO_ONLINE` the
backoff is capped at `_consumeEndTime`, and catch-up gives up at the stuck
offset and downloads. In `CATCHING_UP` retries continue until `stop()`, and the
thread exits without advancing.
--
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]
