anshul98ks123 commented on code in PR #20137:
URL: https://github.com/apache/pinot/pull/20137#discussion_r4457439579
##########
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:
Fixed in c9e9e88a15: a failure that comes a minute or more after the
previous one, with no message being retried, is now reported with its cause and
a segment error, separately from the retry budget. Your 203/210 case is a test
(`testFailureLongAfterThePreviousOneIsReportedAgainWithNoRetryLimit`), which
fails on 38cf4944af.
--
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]
