(kafka) branch 4.3 updated: KAFKA-20808: StreamThread dies with fatal InvalidStateStoreException after TaskCorruptedException (#22855)

[mj...@apache.org]

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

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


The following commit(s) were added to refs/heads/4.3 by this push:
     new 022591e5af3 KAFKA-20808: StreamThread dies with fatal InvalidStateStoreException after TaskCorruptedException (#22855)
022591e5af3 is described below

commit 022591e5af3fbc7b315d374c079e490c1c222e55
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Fri Jul 17 12:38:55 2026 -0700

    KAFKA-20808: StreamThread dies with fatal InvalidStateStoreException after TaskCorruptedException (#22855)
    
    This PR ensures that an unsuccessfully opened segment is remove from the
    in-memory cache for open segments.
    
    Reviewers: Nick Telford <ni...@gmail.com>, Bill Bejeck
     <bb...@apache.org>
---
 .../streams/state/internals/AbstractSegments.java  |  7 ++++++-
 .../state/internals/AbstractSegmentsTest.java      | 24 ++++++++++++++++++++++
 2 files changed, 30 insertions(+), 1 deletion(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractSegments.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractSegments.java
index a46291175e2..8379230e02a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractSegments.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractSegments.java
@@ -96,7 +96,12 @@ public abstract class AbstractSegments<S extends Segment> implements Segments<S>
                 throw new IllegalStateException(newSegment.getClass().getSimpleName() + " already exists. Possible concurrent access.");
             }
 
-            openSegmentDB(newSegment, context);
+            try {
+                openSegmentDB(newSegment, context);
+            } catch (final Exception openException) {
+                segments.remove(segmentId);
+                throw openException;
+            }
             return newSegment;
         }
     }
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractSegmentsTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractSegmentsTest.java
index d729d07f181..df1c52df7c0 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractSegmentsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractSegmentsTest.java
@@ -112,6 +112,30 @@ abstract class AbstractSegmentsTest<S extends Segments> {
         assertEquals("Tasks [0_0] are corrupted and hence need to be re-initialized", thrown.getMessage());
     }
 
+    @Test
+    public void shouldNotRetainFailedSegmentSoRecoveryStaysRecoverable() throws Exception {
+        context = getEOSProcessorContext();
+        segments = getSegments();
+        segments.openExisting(context, 0L);
+        final Segment segment = segments.getOrCreateSegmentIfLive(0, context, 1L);
+
+        assertTrue(segment.isOpen());
+        segments.close();
+
+        simulateUncleanShutdownForAllSegments();
+        segments = getSegments();
+
+        // The first open fails with a (recoverable) TaskCorruptedException because the on-disk state is dirty.
+        assertThrows(TaskCorruptedException.class, () -> segments.openExisting(context, 0L));
+
+        // A segment that failed to open must not be retained in the segments map. Otherwise a subsequent
+        // open would return the stale, unopened segment instead of retrying to open it, and querying its
+        // committed offset during task re-initialization would then fail with a fatal
+        // InvalidStateStoreException. Re-opening must instead stay recoverable, i.e. keep throwing
+        // TaskCorruptedException until the dirty local state is wiped.
+        assertThrows(TaskCorruptedException.class, () -> segments.openExisting(context, 0L));
+    }
+
     private void simulateUncleanShutdownForAllSegments() throws Exception {
         for (final File dbDir : Objects.requireNonNull(context.stateDir().listFiles())) {
             for (final File storeDir : Objects.requireNonNull(dbDir.listFiles())) {