Copilot commented on code in PR #19086:
URL: https://github.com/apache/pinot/pull/19086#discussion_r3669318828
##########
pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisStreamMetadataProvider.java:
##########
@@ -241,21 +293,68 @@ private boolean
consumedEndOfShard(StreamPartitionMsgOffset startCheckpoint,
// Shard can't be iterated further. We have consumed all the
messages because message count = 0
return true;
}
- // Even though message count = 0, shard can be iterated further.
- // Based on kinesis documentation, there might be more records to be
consumed.
- // So we need to fetch messages again to check if we have reached end
of shard.
- // To prevent an infinite loop (known cases listed in
fetchMessages()), we will limit the number of attempts
- attempts++;
- if (attempts >= 5) {
- LOGGER.warn("Reached max attempts to check if end of shard reached
from checkpoint {}. "
- + " Assuming we have not consumed till end of shard.",
startCheckpoint);
+
+ // Empty non-EOP: AWS may need several GetRecords before
nextShardIterator is null on a closed shard,
+ // or the consumer returned a rate-limit/timeout empty batch (no
iterator advance).
+ // Soft-fail throttles/timeouts: a fetch that burned most of its
timeout is treated as transient and does
+ // not increment the hard empty-probe counter. Hard empties (quick
empty responses) do.
+ boolean likelyTransientEmpty = fetchElapsedMs >= (fetchTimeoutMs * 3L
/ 4L);
+ if (!likelyTransientEmpty) {
+ emptyProbes++;
+ }
Review Comment:
`likelyTransientEmpty` uses `fetchTimeoutMs * 3 / 4` as a threshold, but
when `fetchTimeoutMs` is very small (e.g., 1ms) this evaluates to 0, making
*every* empty batch look "transient" (`fetchElapsedMs >= 0`), so `emptyProbes`
never increments. That can prevent the closed-shard probe from ever reaching
the hard-empty budget under low timeout configs (and can reintroduce the "keep
parent live" behavior). Consider clamping the threshold to at least 1ms.
--
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]
