Re: [PR] Fix Kinesis empty segment commit loop after shard merge [pinot]

["Vamsi-klu (via GitHub)" <gi...@apache.org>]

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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org