Ken Huang created KAFKA-21283:
---------------------------------
Summary: 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: Ken Huang
`StreamThreadTest.tearDown()` directly calls `TaskManager.shutdown(false)`
before requesting `StreamThread` shutdown. For tests that start a real
`StreamThread`, 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:
```text
IllegalStateException: Task 0_0 was not found in the state updater.
```
The exception interrupts cleanup before StateUpdater and TaskExecutor shutdown,
and skips the subsequent StreamThread shutdown request. The remaining
TaskExecutor then causes later tests in the same JVM to fail their thread leak
checks. The observed report contains one initial cleanup failure followed by 31
such failures.
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}
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)
