xiangfu0 commented on code in PR #20137:
URL: https://github.com/apache/pinot/pull/20137#discussion_r4452214605
##########
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:
(MINOR) Kafka batches are byte-charged with `getRecordSerializedSize()`
(key, value and headers), but this charges only `getLength()` (the value).
Retry batches are therefore undercharged under byte-based limits.
`chargeProcessedMessages` also counts only processed messages, so tombstones in
a re-fetched batch never reach `REALTIME_ROWS_FETCHED`. Consider using
`getMetadata().getRecordSerializedSize()` when it is positive.
##########
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:
(MINOR) Any successful decode ends the "outage". During a registry outage,
messages with cached schema ids still decode, so every interleaved failing
message gets a fresh full `decodeRetryMaxDurationMillis` stall. I reproduced it
locally: with limit=1000 and failures at 203/205/207 separated by decodable
messages, the retries went `[203,203,205,205,207,207]` with fetches from 203,
205 and 207, i.e. one window per message. A time-based end to the outage (e.g.
no retriable failure for the limit's duration) would bound the outage as
intended. The same applies to `StatelessRealtimeSegmentWriter:328`.
--
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]
