[
https://issues.apache.org/jira/browse/KAFKA-21283?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ken Huang updated KAFKA-21283:
------------------------------
Description:
`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}
was:
`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}
"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}
> 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
> Priority: Major
>
> `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)
