root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-17553 Fix shutdown race condition in StreamThreadTest (#17191) 

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
This commit is contained in:
David Arthur 2024-09-19 03:17:25 -04:00 committed by GitHub
parent 8f5cf9968f
commit 31d395163e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 15 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutor.java
View File

@ -116,7 +116,7 @@ public class DefaultTaskExecutor implements TaskExecutor {
if (currentTask == null) {
try {
taskManager.awaitProcessableTasks();
taskManager.awaitProcessableTasks(shutdownRequested::get);
} catch (final InterruptedException ignored) {
// Can be ignored, the cause of the interrupted will be handled in the event loop
}

13
streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskManager.java
View File

@ -125,7 +125,7 @@ public class DefaultTaskManager implements TaskManager {
}
@Override
public void awaitProcessableTasks() throws InterruptedException {
public void awaitProcessableTasks(final Supplier<Boolean> isShuttingDown) throws InterruptedException {
final boolean interrupted = returnWithTasksLocked(() -> {
for (final Task task : tasks.activeTasks()) {
if (!assignedTasks.containsKey(task.id()) &&
@ -138,8 +138,15 @@ public class DefaultTaskManager implements TaskManager {
}
}
try {
log.debug("Await blocking");
tasksCondition.await();
// We re-check the shutdownRequest atomic boolean to avoid a race condition. If this thread was
// previously interrupted while awaiting tasksCondition, it is possible to miss the signalAll that
// is called during shutdown. If this happens, we end up blocking in the await forever.
if (!isShuttingDown.get()) {
log.debug("Await blocking");
tasksCondition.await();
} else {
log.debug("Not awaiting since shutdown was requested");
}
} catch (final InterruptedException ignored) {
// we interrupt the thread for shut down and pause.
// we can ignore this exception.

3
streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/TaskManager.java
View File

@ -25,6 +25,7 @@ import org.apache.kafka.streams.processor.internals.StreamTask;
import java.time.Duration;
import java.util.Map;
import java.util.Set;
import java.util.function.Supplier;
public interface TaskManager {
@ -136,7 +137,7 @@ public interface TaskManager {
/**
* Blocks until unassigned processable tasks may be available.
*/
void awaitProcessableTasks() throws InterruptedException;
void awaitProcessableTasks(Supplier<Boolean> isShuttingDown) throws InterruptedException;
/**
* Starts all threads associated with this task manager.

2
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
View File

@ -95,7 +95,6 @@ import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
@ -173,7 +172,6 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@Timeout(300)
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.STRICT_STUBS)
public class StreamThreadTest {

2
streams/src/test/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskExecutorTest.java
View File

@ -118,7 +118,7 @@ public class DefaultTaskExecutorTest {
taskExecutor.start();
verify(taskManager, timeout(VERIFICATION_TIMEOUT).atLeastOnce()).awaitProcessableTasks();
verify(taskManager, timeout(VERIFICATION_TIMEOUT).atLeastOnce()).awaitProcessableTasks(any());
}
@Test

2
streams/src/test/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskManagerTest.java
View File

@ -115,7 +115,7 @@ public class DefaultTaskManagerTest {
public void run() {
while (!shutdownRequested.get()) {
try {
taskManager.awaitProcessableTasks();
taskManager.awaitProcessableTasks(shutdownRequested::get);
} catch (final InterruptedException ignored) {
}
awaitDone.countDown();