xiangfu0 commented on code in PR #20020:
URL: https://github.com/apache/pinot/pull/20020#discussion_r4450556933
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -139,18 +151,15 @@ public synchronized KafkaMessageBatch
fetchMessages(StreamPartitionMsgOffset sta
}
lastMessageMetadata = messageMetadata;
}
- } else if (_isReadCommitted) {
- // No records returned. With read_committed isolation, the underlying
KafkaConsumer's
- // internal position may still have advanced past offsets filtered out
as aborted
- // transactional records, so snap _nextReadOffset to the consumer's
actual position.
- // RealtimeSegmentDataManager doesn't advance _currentOffset on empty
batches and will
- // call us again with the same startOffset; without this update the
seek-check would
- // re-seek to startOffset and undo the consumer's progress through the
aborted region.
- //
- // For read_uncommitted (the default), empty polls can never advance the
internal
- // position past records that didn't exist, so we skip the position()
call in the
- // hot path. The call is bounded by the same timeout as poll() so a sick
broker
- // can't stall this loop unbounded.
+ } else {
Review Comment:
Done in dbe385d190. This PR is now the two rate limits, the writer's loss
check, and the flush from the first thread. Both consumers' empty-poll branch
is back to master's (`position()` only under `read_committed`), and the
writer's empty-batch handling is master's too. #20024, which already replaces
both with the validated-position version, carries them now.
I adjusted the tests that pinned the moved behavior:
- the empty-poll reset test now expects the requested offset as the next one,
- the `read_uncommitted` seek test moves to #20024,
- the low-traffic integration scenario does a plain empty fetch at the
marker instead of waiting for the reported position.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
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]
