You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Kuan Po Tseng (Jira)" <ji...@apache.org> on 2026/10/15 07:38:00 UTC
[jira] [Commented] (KAFKA-21283) Fix StreamThreadTest teardown race causing thread leaks and cascading test failures
[ https://issues.apache.org/jira/browse/KAFKA-21283?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18127389#comment-18127389 ]
Kuan Po Tseng commented on KAFKA-21283:
---------------------------------------
Happy to work on this issue, thanks!
> Fix StreamThreadTest teardown race causing thread leaks and cascading test failures
> -----------------------------------------------------------------------------------
>
> Key: KAFKA-21283
> URL: https://issues.apache.org/jira/browse/KAFKA-21283
> Project: Kafka
> Issue Type: Test
> Reporter: Ken Huang
> Assignee: Kuan Po Tseng
> Priority: Major
> Attachments: junit-reports-17-noflaky-nonew.zip
>
>
> `StreamThreadTest.tearDown()` directly calls `TaskManager.shutdown(false)` before requesting `StreamThread` shutdown. This can race with the thread while it is draining restored tasks from `StateUpdater`.
> In `shouldRecoverFromInvalidOffsetExceptionOnRestoreAndFinishRestore(true)`, this can cause shutdown to request removal of a task that the `StreamThread` has already taken, resulting in:
> {code:java}
> IllegalStateException: Task 0_0 was not found in the state updater. {code}
> The exception interrupts cleanup before StateUpdater and TaskExecutor shutdown, and skips the subsequent StreamThread shutdown request. The remaining TaskExecutor then causes later tests to fail their thread leak checks.
> Update teardown to request shutdown and join started StreamThreads, allowing them to clean up their TaskManager through their normal shutdown path. Preserve manual TaskManager cleanup for threads that were never started but whose Streams state was advanced by the test.
> {code:java}
> "shouldRecoverFromInvalidOffsetExceptionOnRestoreAndFinishRestore(boolean).processingThreadsEnabled=true"
> java.lang.IllegalStateException: Task 0_0 was not found in the state updater. This indicates a bug. Please report at https://issues.apache.org/jira/projects/KAFKA/issues or to the dev-mailing list (https://kafka.apache.org/contact).
> at org.apache.kafka.streams.processor.internals.TaskManager.waitForFuture(TaskManager.java:730)
> at org.apache.kafka.streams.processor.internals.TaskManager.lambda$addToTasksToClose$12(TaskManager.java:699)
> at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
> at java.base/java.util.Iterator.forEachRemaining(Iterator.java:133)
> at java.base/java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1845)
> at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
> at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
> at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
> at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
> at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
> at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596)
> at org.apache.kafka.streams.processor.internals.TaskManager.addToTasksToClose(TaskManager.java:701)
> at org.apache.kafka.streams.processor.internals.TaskManager.shutdownStateUpdater(TaskManager.java:1541)
> at org.apache.kafka.streams.processor.internals.TaskManager.shutdown(TaskManager.java:1477)
> at org.apache.kafka.streams.processor.internals.StreamThreadTest.tearDown(StreamThreadTest.java:228)
> at java.base/java.lang.reflect.Method.invoke(Method.java:569)
> at java.base/java.util.Optional.ifPresent(Optional.java:178){code}
> {code:java}
> "shouldAlwaysUpdateTasksMetadataAfterChangingState(boolean).processingThreadsEnabled=true"
> org.opentest4j.AssertionFailedError: expected: <true> but was: <false>
> at org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:151)
> at org.junit.jupiter.api.AssertionFailureBuilder.buildAndThrow(AssertionFailureBuilder.java:132)
> at org.junit.jupiter.api.AssertTrue.failNotTrue(AssertTrue.java:63)
> at org.junit.jupiter.api.AssertTrue.assertTrue(AssertTrue.java:36)
> at org.junit.jupiter.api.AssertTrue.assertTrue(AssertTrue.java:31)
> at org.junit.jupiter.api.Assertions.assertTrue(Assertions.java:183)
> at org.apache.kafka.streams.processor.internals.StreamThreadTest.tearDown(StreamThreadTest.java:234)
> at java.base/java.lang.reflect.Method.invoke(Method.java:569)
> at java.base/java.util.Optional.ifPresent(Optional.java:178)
> at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
> at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
> at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
> at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
> at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
> at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:992)
> at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) {code}
>
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)