[
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)
