anshul98ks123 commented on code in PR #20137:
URL: https://github.com/apache/pinot/pull/20137#discussion_r4452343601
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -663,6 +752,30 @@ 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());
+ decodeRetryNeeded = true;
+ break;
+ }
+ if (_numDecodeRetries > 0) {
+ if (retriable) {
+ _segmentLogger.warn("Giving up retrying the message at offset: {}
after {} retriable decode failures in "
+ + "{}ms. Until a message decodes, retriable decode errors
are handled like any other decode error",
+ offset, _numDecodeRetries, now() - _decodeRetryStartTimeMs);
+ _decodeRetryOutageExceeded = true;
+ } else {
+ _segmentLogger.info("Resumed decoding at offset: {} after {}
retriable decode failures", offset,
+ _numDecodeRetries);
+ }
+ _numDecodeRetries = 0;
+ }
+ if (_decodeRetryOutageExceeded && decodedRow.getException() == null) {
+ _segmentLogger.info("Decoded the message at offset: {}, retrying
retriable decode errors again", offset);
+ _decodeRetryOutageExceeded = false;
Review Comment:
Fixed in 28960113d6: an outage now ends only after the limit's duration
passes with no retriable failure, so decodes in between don't reset it; your
203/205/207 case is a test in both classes (only 203 retries, 205 and 207 drop
at once).
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -353,6 +365,54 @@ public void deleteSegmentFile() {
private volatile boolean _stopReasonPrinted = false;
+ /// Outcome of [#processStreamEvents].
+ private enum BatchOutcome {
+ /// Every message of the batch was processed.
+ PROCESSED,
+ /// Processing stopped early because consumption is stopping or an end
criterion was reached.
+ STOPPED_EARLY,
+ /// Processing stopped at a message whose decoder threw
`RetriableDecodeException`. [#_currentOffset] was not moved
+ /// past it, so the message is fetched and decoded again.
+ DECODE_RETRY_NEEDED
+ }
+
+ /// The first messages of a batch, which the rate limiters are charged for
when the batch was fetched again to retry a
+ /// decode. Its size is the size of the message values.
+ private static final class ProcessedMessages<T> implements MessageBatch<T> {
+ private final MessageBatch<T> _messageBatch;
+ private final int _numMessages;
+
+ ProcessedMessages(MessageBatch<T> messageBatch, int numMessages) {
+ _messageBatch = messageBatch;
+ _numMessages = numMessages;
+ }
+
+ @Override
+ public int getMessageCount() {
+ return _numMessages;
+ }
+
+ @Override
+ public StreamMessage<T> getStreamMessage(int index) {
+ Preconditions.checkElementIndex(index, _numMessages);
+ return _messageBatch.getStreamMessage(index);
+ }
+
+ @Override
+ public StreamPartitionMsgOffset getOffsetOfNextBatch() {
+ return getStreamMessage(_numMessages - 1).getMetadata().getNextOffset();
+ }
+
+ @Override
+ public long getSizeInBytes() {
+ long sizeInBytes = 0;
+ for (int i = 0; i < _numMessages; i++) {
+ sizeInBytes += _messageBatch.getStreamMessage(i).getLength();
Review Comment:
Fixed in 38cf4944af: retried messages are sized by
`getRecordSerializedSize()` when positive, else `getLength()`, and a fully
processed retry batch charges its unfiltered count like a normal batch (test
adds a tombstone and headers).
--
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]
