Fix actual seek-on-empty-poll wedge in KafkaPartitionLevelConsumer

Spotted a bug in my own previous attempt while staring at the unchanged CI
failure: setting _lastFetchedOffset = startOffset - 1 in the empty-poll
branch does not skip the re-seek when startOffset == 0 (= -1, which still
trips the _lastFetchedOffset < 0 clause of the seek-check). That clause
was the original intent's "have we ever fetched" sentinel and a sentinel
value of -1 is indistinguishable from "we haven't fetched yet" and "we
fetched at offset 0".

Replace with an explicit _lastSeekedStartOffset (Long.MIN_VALUE initially)
that tracks whether we've already issued a seek for the caller's
startOffset. The seek-check is now:
  firstSeekForThisOffset = _lastSeekedStartOffset != startOffset
  if (firstSeekForThisOffset || _lastFetchedOffset != startOffset - 1)
The first time the caller asks for startOffset = N, we seek and record N.
On subsequent calls with the same startOffset we do NOT re-seek, so the
consumer's internal fetch session can advance through aborted records on
later polls instead of being reset to startOffset every time.

This is the part that the previous fix missed -- _lastFetchedOffset = -1
+ the existing < 0 check kept re-seeking even after we marked the offset.

Mirrored to both pinot-kafka-3.0 and pinot-kafka-4.0.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: xiangfu0

pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java

@@ -47,9 +47,14 @@ public class KafkaPartitionLevelConsumer extends KafkaPartitionLevelConnectionHa
   private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPartitionLevelConsumer.class);
 
   private long _lastFetchedOffset = -1;
-  // Diagnostic counters: number of consecutive fetch calls that returned no records, and
-  // the last log of the empty-poll path. Logged at WARN/ERROR every ~50 empty polls so a
-  // long stall surfaces in CI surefire output without spamming the success path.
+  // True once the consumer has been positioned (via seek or assign) for the current
+  // startOffset and has had at least one poll attempt. Tracked separately from
+  // _lastFetchedOffset because, with read_committed isolation, the very first poll can
+  // legitimately return zero records (all records were aborted) and we must NOT re-seek
+  // on the next call -- doing so would undo the consumer's internal advance through the
+  // aborted region and wedge us forever at startOffset = 0.
+  private long _lastSeekedStartOffset = Long.MIN_VALUE;
+  // Diagnostic counter: consecutive fetch calls that returned no records.
   private int _consecutiveEmptyPolls = 0;
 
   public KafkaPartitionLevelConsumer(String clientId, StreamConfig streamConfig, int partition) {
@@ -67,11 +72,19 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
     if (LOGGER.isDebugEnabled()) {
       LOGGER.debug("Polling partition: {}, startOffset: {}, timeout: {}ms", _topicPartition, startOffset, timeoutMs);
     }
-    if (_lastFetchedOffset < 0 || _lastFetchedOffset != startOffset - 1) {
+    // Seek if (a) we've never positioned the consumer, OR (b) the caller's startOffset
+    // moved off the position we last tracked. _lastFetchedOffset < 0 alone is NOT a
+    // sufficient seek trigger because, with read_committed, an empty first poll keeps
+    // _lastFetchedOffset at its initial -1 even though the consumer's internal position
+    // has moved past aborted records; re-seeking on every empty poll would undo that
+    // progress and wedge consumption forever at startOffset.
+    boolean firstSeekForThisOffset = _lastSeekedStartOffset != startOffset;
+    if (firstSeekForThisOffset || _lastFetchedOffset != startOffset - 1) {
       if (LOGGER.isDebugEnabled()) {
         LOGGER.debug("Seeking to offset: {}", startOffset);
       }
       _consumer.seek(_topicPartition, startOffset);
+      _lastSeekedStartOffset = startOffset;
     }
 
     ConsumerRecords<Bytes, Bytes> consumerRecords = _consumer.poll(Duration.ofMillis(timeoutMs));

pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java

@@ -47,8 +47,14 @@ public class KafkaPartitionLevelConsumer extends KafkaPartitionLevelConnectionHa
   private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPartitionLevelConsumer.class);
 
   private long _lastFetchedOffset = -1;
-  // Diagnostic counter: consecutive fetch calls that returned no records. Logged at WARN
-  // every ~50 empty polls so a long stall surfaces in CI surefire output without spam.
+  // True once the consumer has been positioned (via seek or assign) for the current
+  // startOffset and has had at least one poll attempt. Tracked separately from
+  // _lastFetchedOffset because, with read_committed isolation, the very first poll can
+  // legitimately return zero records (all records were aborted) and we must NOT re-seek
+  // on the next call -- doing so would undo the consumer's internal advance through the
+  // aborted region and wedge us forever at startOffset = 0.
+  private long _lastSeekedStartOffset = Long.MIN_VALUE;
+  // Diagnostic counter: consecutive fetch calls that returned no records.
   private int _consecutiveEmptyPolls = 0;
 
   public KafkaPartitionLevelConsumer(String clientId, StreamConfig streamConfig, int partition) {
@@ -66,11 +72,19 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
     if (LOGGER.isDebugEnabled()) {
       LOGGER.debug("Polling partition: {}, startOffset: {}, timeout: {}ms", _topicPartition, startOffset, timeoutMs);
     }
-    if (_lastFetchedOffset < 0 || _lastFetchedOffset != startOffset - 1) {
+    // Seek if (a) we've never positioned the consumer for this startOffset, OR (b) the
+    // caller's startOffset moved off the position we last tracked. _lastFetchedOffset < 0
+    // alone is NOT a sufficient seek trigger because, with read_committed, an empty
+    // first poll keeps _lastFetchedOffset at its initial -1 even though the consumer's
+    // internal position has moved past aborted records; re-seeking on every empty poll
+    // would undo that progress and wedge consumption forever at startOffset.
+    boolean firstSeekForThisOffset = _lastSeekedStartOffset != startOffset;
+    if (firstSeekForThisOffset || _lastFetchedOffset != startOffset - 1) {
       if (LOGGER.isDebugEnabled()) {
         LOGGER.debug("Seeking to offset: {}", startOffset);
       }
       _consumer.seek(_topicPartition, startOffset);
+      _lastSeekedStartOffset = startOffset;
     }
 
     ConsumerRecords<Bytes, Bytes> consumerRecords = _consumer.poll(Duration.ofMillis(timeoutMs));