(kafka) branch trunk updated: KAFKA-21159: Clear next offsets to consume on task close to avoid stale commit after revive (#23584)

[bb...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

bbejeck pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 7a6029541a0 KAFKA-21159: Clear next offsets to consume on task close to avoid stale commit after revive (#23584)
7a6029541a0 is described below

commit 7a6029541a0538d8529ad69e2bce6012cc1fb229
Author: Bill Bejeck <bb...@apache.org>
AuthorDate: Tue Sep 29 10:03:06 2026 -0400

    KAFKA-21159: Clear next offsets to consume on task close to avoid stale commit after revive (#23584)
    
    A revived active task (`closeDirtyAndRevive`) kept
    `nextOffsetsAndMetadataToBeConsumed`  from before the corruption. After
    the consumer is seeked back to its committed offset, a  commit driven by
    processor metadata (`needsCommit`, which commits all input partitions)
    could fall back to that stale next offset and commit past records the
    task hasn't  reprocessed.
    
    Fix: clear `nextOffsetsAndMetadataToBeConsumed` in `StreamTask.close()`,
    alongside  `consumedOffsets`, so a revived task has no stale fallback
    for a partition it hasn't  re-read.
    
    Testing: added
    
    `StreamTaskTest.shouldNotCommitStaleNextOffsetForPartitionNotReReadAfterRevive`,
    which reproduces the revive scenario and fails without the fix. Full
    `StreamTaskTest` passes.
    
    Reviewers: Rahul <ra...@gmail.com>, Lucas Brutschy
     <lb...@confluent.io>
---
 .../streams/processor/internals/StreamTask.java    |  2 +
 .../processor/internals/StreamTaskTest.java        | 46 ++++++++++++++++++++++
 2 files changed, 48 insertions(+)

diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
index 39fd0e56b4e..22b8f9bfff1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -713,6 +713,8 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator,
         closeTaskSensor.record();
         partitionsToResume.clear();
         consumedOffsets.clear();
+        // Clear so a revived task has no stale fallback for a partition it has not re-read.
+        nextOffsetsAndMetadataToBeConsumed.clear();
 
         transitionTo(State.CLOSED);
     }
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
index b0bd2a737bd..ad528c29744 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
@@ -2805,6 +2805,52 @@ public class StreamTaskTest {
         assertFalse(task.commitRequested());
     }
 
+    @Test
+    public void shouldNotCommitStaleNextOffsetForPartitionNotReReadAfterRevive() {
+        // A revived task must not commit a stale "next offset to consume" for a partition it has not re-read.
+        when(stateManager.taskId()).thenReturn(taskId);
+        when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+        // idle disabled (-1) so the revived task can process partition2 without waiting on empty partition1
+        task = createStatelessTask(createConfig(AT_LEAST_ONCE, "-1"));
+
+        // committed offset partition1 will be seeked back to when the revived task restores
+        consumer.commitSync(Map.of(partition1, new OffsetAndMetadata(5L)));
+
+        task.initializeIfNeeded();
+        task.completeRestoration(noOpResetter -> { });
+
+        // a poll advanced partition1 past its committed offset before the corruption
+        task.updateNextOffsets(partition1, new OffsetAndMetadata(10L, Optional.of(0), ""));
+
+        // closeDirtyAndRevive: suspend, close dirty, mark inputs for offset reset, revive
+        task.suspend();
+        task.closeDirty();
+        task.addPartitionsForOffsetReset(Set.of(partition1));
+        task.revive();
+
+        // restoration seeks partition1 back to its committed offset; partition1 is now NOT re-read
+        task.initializeIfNeeded();
+        task.completeRestoration(noOpResetter -> { });
+        assertEquals(5L, consumer.position(partition1));
+
+        // the revived task re-reads partition2 only, and processor metadata forces a commit of all inputs
+        task.addRecords(partition2, singletonList(getConsumerRecordWithOffsetAsTimestampWithLeaderEpoch(partition2, 0L, 0)));
+        task.process(0L);
+        task.updateNextOffsets(partition2, new OffsetAndMetadata(1L, Optional.of(0), ""));
+        processorStreamTime.mockProcessor.addProcessorMetadata("key1", 100L);
+
+        assertTrue(task.commitNeeded());
+        final Map<TopicPartition, OffsetAndMetadata> committed = task.prepareCommit(true);
+
+        // partition1 must be left out: committing its stale next offset (10) would skip records 5..9
+        assertFalse(
+            committed.containsKey(partition1),
+            "revived task committed a stale next offset for a partition it did not re-read"
+        );
+        // partition2, which was re-read, is still committed
+        assertTrue(committed.containsKey(partition2));
+    }
+
     @Test
     public void closeShouldBeIdempotent() {
         when(stateManager.taskId()).thenReturn(taskId);