Vamsi-klu commented on code in PR #19086:
URL: https://github.com/apache/pinot/pull/19086#discussion_r3725702428
##########
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:
Good catch, thanks. One clarification: the loop cannot spin indefinitely
since maxTotalAttempts (20) bounds it, but you are right that with
fetchTimeoutMs clamped to 1 the threshold evaluates to 0, so every empty probe
is classified transient, emptyProbes stays at 0, and both exit paths then
return false, which reintroduces the keep parent live behavior for sub 2ms
timeout configs. The same edge can also hit the final probe of a normal config
when remainingMs shrinks the effective timeout near the deadline. I will clamp
the threshold to at least 1ms, i.e. Math.max(1L, fetchTimeoutMs * 3L / 4L), and
add a unit test for fetchTimeoutMs = 1.
--
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]
