xiangfu0 commented on code in PR #20137:
URL: https://github.com/apache/pinot/pull/20137#discussion_r4471369295
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -776,7 +899,111 @@ 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;
+ }
+
+ @VisibleForTesting
+ void throttle(MessageBatch<?> messageBatch) {
+ _partitionRateLimiter.throttle(messageBatch);
+ _serverRateLimiter.throttle(messageBatch);
+ }
+
+ /// Charges the rate limiters and the fetched rows meter for the first
`numMessages` messages of a batch fetched again
+ /// to retry a decode, which stopped processing after them. Charging after
processing keeps the rate, since the rate
+ /// limiters let a caller acquire permits ahead of the rate and make the
next caller wait instead.
+ /// The fetched rows meter misses the messages filtered out of the batch
(e.g. tombstones) before the stopping point,
+ /// since a batch does not tell where they are.
+ private void chargeProcessedMessages(MessageBatch<?> messageBatch, int
numMessages) {
+ if (numMessages > 0) {
+ throttle(new ProcessedMessages<>(messageBatch, numMessages));
+ _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.REALTIME_ROWS_FETCHED, numMessages);
+ }
+ }
+
+ /// Handles a retriable decode failure of the message at `offset`, and
returns whether to fetch and decode it again.
+ /// Otherwise, it is handled like any other decode failure.
+ ///
+ /// Retriable failures belong to a decode outage, e.g. of a schema registry.
It starts at a retriable failure, and
+ /// lasts while a message is being retried and while retriable failures keep
happening within
+ /// [StreamConfig#getDecodeRetryMaxDurationMillis()] of the previous one,
whether or not messages decode in between.
+ /// Messages are retried until the outage has lasted that long; after that,
its retriable failures are not retried.
+ private boolean shouldRetryDecode(StreamPartitionMsgOffset offset, Exception
cause) {
+ long nowMs = now();
+ long maxDurationMillis = _streamConfig.getDecodeRetryMaxDurationMillis();
+ long sinceLastFailureMs = _inDecodeOutage ? nowMs -
_decodeOutageLastFailureTimeMs : Long.MAX_VALUE;
+ boolean outageStarts = !_inDecodeOutage || (_numDecodeRetries == 0 &&
sinceLastFailureMs >= maxDurationMillis);
+ if (outageStarts) {
+ _inDecodeOutage = true;
+ _decodeOutageStartTimeMs = nowMs;
+ _decodeOutageRetryBudgetSpent = false;
+ }
+ _decodeOutageLastFailureTimeMs = nowMs;
+ long outageDurationMs = nowMs - _decodeOutageStartTimeMs;
+ if (outageDurationMs >= maxDurationMillis) {
+ if (maxDurationMillis > 0 && !_decodeOutageRetryBudgetSpent) {
+ _segmentLogger.warn("Retriable decode errors have lasted {}ms, giving
up retrying the message at offset: {}. "
+ + "Until none happens for {}ms, retriable decode errors are
handled like any other decode error",
+ outageDurationMs, offset, maxDurationMillis);
+ _decodeOutageRetryBudgetSpent = true;
+ }
+ _numDecodeRetries = 0;
+ return false;
+ }
+
+ // Retry the message, counting the retry. A failure that starts an outage,
or that comes a minute or more after the
+ // previous one with no message being retried, is logged at WARN with its
cause and recorded as a segment error.
+ // The rest are logged at WARN at most once a minute. Reporting does not
follow the retry budget, so that with no
+ // retry limit, where an outage never ends, a failure long after the last
one is still reported with its cause.
+ _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.REALTIME_DECODE_RETRIES, 1L);
+ boolean reportFailure =
+ outageStarts || (_numDecodeRetries == 0 && sinceLastFailureMs >=
DECODE_RETRY_WARN_INTERVAL_MS);
+ if (_numDecodeRetries++ == 0) {
+ _decodeRetryBackoffMs =
Math.min(StreamConfig.INITIAL_DECODE_RETRY_BACKOFF_MILLIS,
+ _streamConfig.getDecodeRetryMaxBackoffMillis());
+ }
+ if (reportFailure) {
+ _decodeOutageLastWarnTimeMs = nowMs;
+ 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 - _decodeOutageLastWarnTimeMs >=
DECODE_RETRY_WARN_INTERVAL_MS) {
+ _decodeOutageLastWarnTimeMs = nowMs;
+ _segmentLogger.warn("Still retrying: retriable decode error at offset:
{}, {}ms into the outage", offset,
+ outageDurationMs, cause);
+ } else if (_segmentLogger.isDebugEnabled()) {
+ _segmentLogger.debug("Retriable decode error at offset: {} (failure {}
of the message)", offset,
+ _numDecodeRetries, cause);
+ }
+ return true;
+ }
+
+ /// 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. It is skipped once a force commit, which pausing
consumption also sends, ends the segment there.
+ /// [#stop()] interrupts the sleep.
+ private void backOffBeforeDecodeRetry()
+ throws InterruptedException {
+ if (_shouldStop || (_state == State.INITIAL_CONSUMING &&
_forceCommitMessageReceived)) {
Review Comment:
(MINOR) This only helps when the force commit arrives before the backoff
starts. `forceCommit()` just sets the flag, so a commit that lands mid-sleep
still waits up to `decodeRetryMaxBackoffMillis` (10s by default). I reproduced
it locally: I started a real backoff sleep and called `forceCommit()`. Two
seconds later the consumer was still `TIMED_WAITING` in the first backoff.
Since the backoff takes up most of the time while a message is stuck, most
force commits will land mid-sleep. Consider sleeping in short slices that check
the flag, or having `forceCommit()` wake the consumer.
--
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]
