Vamsi-klu commented on code in PR #19086:
URL: https://github.com/apache/pinot/pull/19086#discussion_r3755546219
##########
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:
Clamped in `6669b54`, the threshold is `Math.max(1L, fetchTimeoutMs * 3L /
4L)` now.
Added two regression tests, one with `fetchTimeoutMs = 1` and one where
`remainingMs` shrinks the effective timeout near the deadline. Both use the
existing overridable clock hook and both run against a closed shard so child
admission is covered too. Without the clamp the first test keeps the parent
live for the full attempt budget, which is exactly the behavior you described.
--
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]
