xiangfu0 commented on code in PR #20137:
URL: https://github.com/apache/pinot/pull/20137#discussion_r4453273481
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -776,7 +898,106 @@ 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();
+ boolean outageStarts = !_inDecodeOutage
Review Comment:
(MINOR, observability) With the unlimited default, `nowMs -
_decodeOutageLastFailureTimeMs >= Long.MAX_VALUE` is never true, so a segment's
first outage never ends. I reproduced this locally with the default config:
offset 203 failed once and recovered, then offset 210 failed 5 minutes later.
That second failure logged `Still retrying: ... 300000ms into the outage`, and
`addSegmentError` was called only once. Retrying still worked, but the
cause-bearing WARN and the segment error are lost for every later outage.
Consider ending the logging episode after a short quiet window (e.g.
`DECODE_RETRY_WARN_INTERVAL_MS`) separately from the retry budget.
--
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]
