You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/05/19 10:22:04 UTC
(kafka) branch trunk updated: KAFKA-16789 Fix thread leak detection for event handler threads (#15984)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 412b05df001 KAFKA-16789 Fix thread leak detection for event handler threads (#15984)
412b05df001 is described below
commit 412b05df001df399a873b74093e26b1f6f70ef44
Author: Gaurav Narula <ga...@apple.com>
AuthorDate: Sun May 19 11:21:56 2024 +0100
KAFKA-16789 Fix thread leak detection for event handler threads (#15984)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../test/scala/unit/kafka/utils/TestUtils.scala | 4 +-
.../apache/kafka/controller/QuorumController.java | 2 -
.../org/apache/kafka/queue/KafkaEventQueue.java | 5 +-
.../apache/kafka/queue/KafkaEventQueueTest.java | 445 +++++++++++----------
4 files changed, 234 insertions(+), 222 deletions(-)
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index 9f58920d8f8..ab1b3738cd5 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -51,11 +51,11 @@ import org.apache.kafka.common.security.auth.{KafkaPrincipal, KafkaPrincipalSerd
import org.apache.kafka.common.serialization._
import org.apache.kafka.common.utils.Utils.formatAddress
import org.apache.kafka.common.utils.{Time, Utils}
-import org.apache.kafka.controller.QuorumController
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
import org.apache.kafka.coordinator.transaction.TransactionLogConfigs
import org.apache.kafka.metadata.properties.MetaProperties
import org.apache.kafka.network.SocketServerConfigs
+import org.apache.kafka.queue.KafkaEventQueue
import org.apache.kafka.raft.QuorumConfig
import org.apache.kafka.server.authorizer.{AuthorizableRequestContext, Authorizer => JAuthorizer}
import org.apache.kafka.server.common.{ApiMessageAndVersion, MetadataVersion}
@@ -1872,7 +1872,7 @@ object TestUtils extends Logging {
AdminClientUnitTestEnv.kafkaAdminClientNetworkThreadPrefix(),
AbstractCoordinator.HEARTBEAT_THREAD_PREFIX,
QuorumTestHarness.ZkClientEventThreadSuffix,
- QuorumController.CONTROLLER_THREAD_SUFFIX,
+ KafkaEventQueue.EVENT_HANDLER_THREAD_SUFFIX,
ClientMetricsManager.CLIENT_METRICS_REAPER_THREAD_NAME,
SystemTimer.SYSTEM_TIMER_THREAD_PREFIX,
)
diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
index 789c369c9b2..9d186d83d3f 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
@@ -499,8 +499,6 @@ public final class QuorumController implements Controller {
}
}
- public static final String CONTROLLER_THREAD_SUFFIX = "QuorumControllerEventHandler";
-
private OptionalInt latestController() {
return raftClient.leaderAndEpoch().leaderId();
}
diff --git a/server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java b/server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java
index 972ceb1e934..3611359f72a 100644
--- a/server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java
+++ b/server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java
@@ -36,6 +36,9 @@ import org.slf4j.Logger;
public final class KafkaEventQueue implements EventQueue {
+
+ public static final String EVENT_HANDLER_THREAD_SUFFIX = "event-handler";
+
/**
* A context object that wraps events.
*/
@@ -454,7 +457,7 @@ public final class KafkaEventQueue implements EventQueue {
this.lock = new ReentrantLock();
this.log = logContext.logger(KafkaEventQueue.class);
this.eventHandler = new EventHandler();
- this.eventHandlerThread = new KafkaThread(threadNamePrefix + "event-handler",
+ this.eventHandlerThread = new KafkaThread(threadNamePrefix + EVENT_HANDLER_THREAD_SUFFIX,
this.eventHandler, false);
this.shuttingDown = false;
this.interrupted = false;
diff --git a/server-common/src/test/java/org/apache/kafka/queue/KafkaEventQueueTest.java b/server-common/src/test/java/org/apache/kafka/queue/KafkaEventQueueTest.java
index 1fdb9f40e68..41a85f9ca0b 100644
--- a/server-common/src/test/java/org/apache/kafka/queue/KafkaEventQueueTest.java
+++ b/server-common/src/test/java/org/apache/kafka/queue/KafkaEventQueueTest.java
@@ -32,7 +32,10 @@ import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.Timeout;
import static java.util.concurrent.TimeUnit.HOURS;
@@ -65,167 +68,177 @@ public class KafkaEventQueueTest {
}
}
+ private LogContext logContext;
+
+ @AfterAll
+ public static void tearDown() throws InterruptedException {
+ TestUtils.waitForCondition(
+ () -> Thread.getAllStackTraces().keySet().stream()
+ .map(Thread::getName)
+ .noneMatch(t -> t.endsWith(KafkaEventQueue.EVENT_HANDLER_THREAD_SUFFIX)),
+ "Thread leak detected"
+ );
+ }
+
+ @BeforeEach
+ public void setUp(TestInfo testInfo) {
+ logContext = new LogContext("[KafkaEventQueue test=" + testInfo.getDisplayName() + "]");
+ }
+
@Test
public void testCreateAndClose() throws Exception {
KafkaEventQueue queue =
- new KafkaEventQueue(Time.SYSTEM, new LogContext(), "testCreateAndClose");
+ new KafkaEventQueue(Time.SYSTEM, logContext, "testCreateAndClose");
queue.close();
}
@Test
public void testHandleEvents() throws Exception {
- KafkaEventQueue queue =
- new KafkaEventQueue(Time.SYSTEM, new LogContext(), "testHandleEvents");
- AtomicInteger numEventsExecuted = new AtomicInteger(0);
- CompletableFuture<Integer> future1 = new CompletableFuture<>();
- queue.prepend(new FutureEvent<>(future1, () -> {
- assertEquals(1, numEventsExecuted.incrementAndGet());
- return 1;
- }));
- CompletableFuture<Integer> future2 = new CompletableFuture<>();
- queue.appendWithDeadline(Time.SYSTEM.nanoseconds() + TimeUnit.SECONDS.toNanos(60),
- new FutureEvent<>(future2, () -> {
- assertEquals(2, numEventsExecuted.incrementAndGet());
- return 2;
+ try (KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext, "testHandleEvents")) {
+ AtomicInteger numEventsExecuted = new AtomicInteger(0);
+ CompletableFuture<Integer> future1 = new CompletableFuture<>();
+ queue.prepend(new FutureEvent<>(future1, () -> {
+ assertEquals(1, numEventsExecuted.incrementAndGet());
+ return 1;
}));
- CompletableFuture<Integer> future3 = new CompletableFuture<>();
- queue.append(new FutureEvent<>(future3, () -> {
- assertEquals(3, numEventsExecuted.incrementAndGet());
- return 3;
- }));
- assertEquals(Integer.valueOf(1), future1.get());
- assertEquals(Integer.valueOf(3), future3.get());
- assertEquals(Integer.valueOf(2), future2.get());
- CompletableFuture<Integer> future4 = new CompletableFuture<>();
- queue.appendWithDeadline(Time.SYSTEM.nanoseconds() + TimeUnit.SECONDS.toNanos(60),
- new FutureEvent<>(future4, () -> {
- assertEquals(4, numEventsExecuted.incrementAndGet());
- return 4;
+ CompletableFuture<Integer> future2 = new CompletableFuture<>();
+ queue.appendWithDeadline(Time.SYSTEM.nanoseconds() + TimeUnit.SECONDS.toNanos(60),
+ new FutureEvent<>(future2, () -> {
+ assertEquals(2, numEventsExecuted.incrementAndGet());
+ return 2;
+ }));
+ CompletableFuture<Integer> future3 = new CompletableFuture<>();
+ queue.append(new FutureEvent<>(future3, () -> {
+ assertEquals(3, numEventsExecuted.incrementAndGet());
+ return 3;
}));
- future4.get();
- queue.beginShutdown("testHandleEvents");
- queue.close();
+ assertEquals(Integer.valueOf(1), future1.get());
+ assertEquals(Integer.valueOf(3), future3.get());
+ assertEquals(Integer.valueOf(2), future2.get());
+ CompletableFuture<Integer> future4 = new CompletableFuture<>();
+ queue.appendWithDeadline(Time.SYSTEM.nanoseconds() + TimeUnit.SECONDS.toNanos(60),
+ new FutureEvent<>(future4, () -> {
+ assertEquals(4, numEventsExecuted.incrementAndGet());
+ return 4;
+ }));
+ future4.get();
+ }
}
@Test
public void testTimeouts() throws Exception {
- KafkaEventQueue queue =
- new KafkaEventQueue(Time.SYSTEM, new LogContext(), "testTimeouts");
- AtomicInteger numEventsExecuted = new AtomicInteger(0);
- CompletableFuture<Integer> future1 = new CompletableFuture<>();
- queue.append(new FutureEvent<>(future1, () -> {
- assertEquals(1, numEventsExecuted.incrementAndGet());
- return 1;
- }));
- CompletableFuture<Integer> future2 = new CompletableFuture<>();
- queue.append(new FutureEvent<>(future2, () -> {
- assertEquals(2, numEventsExecuted.incrementAndGet());
- Time.SYSTEM.sleep(1);
- return 2;
- }));
- CompletableFuture<Integer> future3 = new CompletableFuture<>();
- queue.appendWithDeadline(Time.SYSTEM.nanoseconds() + 1,
- new FutureEvent<>(future3, () -> {
+ try (KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext, "testTimeouts")) {
+ AtomicInteger numEventsExecuted = new AtomicInteger(0);
+ CompletableFuture<Integer> future1 = new CompletableFuture<>();
+ queue.append(new FutureEvent<>(future1, () -> {
+ assertEquals(1, numEventsExecuted.incrementAndGet());
+ return 1;
+ }));
+ CompletableFuture<Integer> future2 = new CompletableFuture<>();
+ queue.append(new FutureEvent<>(future2, () -> {
+ assertEquals(2, numEventsExecuted.incrementAndGet());
+ Time.SYSTEM.sleep(1);
+ return 2;
+ }));
+ CompletableFuture<Integer> future3 = new CompletableFuture<>();
+ queue.appendWithDeadline(Time.SYSTEM.nanoseconds() + 1,
+ new FutureEvent<>(future3, () -> {
+ numEventsExecuted.incrementAndGet();
+ return 3;
+ }));
+ CompletableFuture<Integer> future4 = new CompletableFuture<>();
+ queue.append(new FutureEvent<>(future4, () -> {
numEventsExecuted.incrementAndGet();
- return 3;
+ return 4;
}));
- CompletableFuture<Integer> future4 = new CompletableFuture<>();
- queue.append(new FutureEvent<>(future4, () -> {
- numEventsExecuted.incrementAndGet();
- return 4;
- }));
- assertEquals(Integer.valueOf(1), future1.get());
- assertEquals(Integer.valueOf(2), future2.get());
- assertEquals(Integer.valueOf(4), future4.get());
- assertEquals(TimeoutException.class,
- assertThrows(ExecutionException.class,
- () -> future3.get()).getCause().getClass());
- queue.close();
- assertEquals(3, numEventsExecuted.get());
+ assertEquals(Integer.valueOf(1), future1.get());
+ assertEquals(Integer.valueOf(2), future2.get());
+ assertEquals(Integer.valueOf(4), future4.get());
+ assertEquals(TimeoutException.class,
+ assertThrows(ExecutionException.class,
+ () -> future3.get()).getCause().getClass());
+ assertEquals(3, numEventsExecuted.get());
+ }
}
@Test
public void testScheduleDeferred() throws Exception {
- KafkaEventQueue queue =
- new KafkaEventQueue(Time.SYSTEM, new LogContext(), "testAppendDeferred");
-
- // Wait for the deferred event to happen after the non-deferred event.
- // It may not happen every time, so we keep trying until it does.
- AtomicLong counter = new AtomicLong(0);
- CompletableFuture<Boolean> future1;
- do {
- counter.addAndGet(1);
- future1 = new CompletableFuture<>();
- queue.scheduleDeferred(null,
- __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + 1000000),
- new FutureEvent<>(future1, () -> counter.get() % 2 == 0));
- CompletableFuture<Long> future2 = new CompletableFuture<>();
- queue.append(new FutureEvent<>(future2, () -> counter.addAndGet(1)));
- future2.get();
- } while (!future1.get());
- queue.close();
+ try (KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext, "testAppendDeferred")) {
+
+ // Wait for the deferred event to happen after the non-deferred event.
+ // It may not happen every time, so we keep trying until it does.
+ AtomicLong counter = new AtomicLong(0);
+ CompletableFuture<Boolean> future1;
+ do {
+ counter.addAndGet(1);
+ future1 = new CompletableFuture<>();
+ queue.scheduleDeferred(null,
+ __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + 1000000),
+ new FutureEvent<>(future1, () -> counter.get() % 2 == 0));
+ CompletableFuture<Long> future2 = new CompletableFuture<>();
+ queue.append(new FutureEvent<>(future2, () -> counter.addAndGet(1)));
+ future2.get();
+ } while (!future1.get());
+ }
}
private final static long ONE_HOUR_NS = TimeUnit.NANOSECONDS.convert(1, HOURS);
@Test
public void testScheduleDeferredWithTagReplacement() throws Exception {
- KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, new LogContext(),
- "testScheduleDeferredWithTagReplacement");
-
- AtomicInteger ai = new AtomicInteger(0);
- CompletableFuture<Integer> future1 = new CompletableFuture<>();
- queue.scheduleDeferred("foo",
- __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + ONE_HOUR_NS),
- new FutureEvent<>(future1, () -> ai.addAndGet(1000)));
- CompletableFuture<Integer> future2 = new CompletableFuture<>();
- queue.scheduleDeferred("foo", prev -> OptionalLong.of(prev.orElse(0) - ONE_HOUR_NS),
- new FutureEvent<>(future2, () -> ai.addAndGet(1)));
- assertFalse(future1.isDone());
- assertEquals(Integer.valueOf(1), future2.get());
- assertEquals(1, ai.get());
- queue.close();
+ try (KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext, "testScheduleDeferredWithTagReplacement")) {
+
+ AtomicInteger ai = new AtomicInteger(0);
+ CompletableFuture<Integer> future1 = new CompletableFuture<>();
+ queue.scheduleDeferred("foo",
+ __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + ONE_HOUR_NS),
+ new FutureEvent<>(future1, () -> ai.addAndGet(1000)));
+ CompletableFuture<Integer> future2 = new CompletableFuture<>();
+ queue.scheduleDeferred("foo", prev -> OptionalLong.of(prev.orElse(0) - ONE_HOUR_NS),
+ new FutureEvent<>(future2, () -> ai.addAndGet(1)));
+ assertFalse(future1.isDone());
+ assertEquals(Integer.valueOf(1), future2.get());
+ assertEquals(1, ai.get());
+ }
}
@Test
public void testDeferredIsQueuedAfterTriggering() throws Exception {
MockTime time = new MockTime(0, 100000, 1);
- KafkaEventQueue queue = new KafkaEventQueue(time, new LogContext(),
- "testDeferredIsQueuedAfterTriggering");
- AtomicInteger count = new AtomicInteger(0);
- List<CompletableFuture<Integer>> futures = Arrays.asList(
- new CompletableFuture<>(),
- new CompletableFuture<>(),
- new CompletableFuture<>());
- queue.scheduleDeferred("foo", __ -> OptionalLong.of(2L),
- new FutureEvent<>(futures.get(0), () -> count.getAndIncrement()));
- queue.append(new FutureEvent<>(futures.get(1), () -> count.getAndAdd(1)));
- assertEquals(Integer.valueOf(0), futures.get(1).get());
- time.sleep(1);
- queue.append(new FutureEvent<>(futures.get(2), () -> count.getAndAdd(1)));
- assertEquals(Integer.valueOf(1), futures.get(0).get());
- assertEquals(Integer.valueOf(2), futures.get(2).get());
- queue.close();
+ try (KafkaEventQueue queue = new KafkaEventQueue(time, logContext, "testDeferredIsQueuedAfterTriggering")) {
+ AtomicInteger count = new AtomicInteger(0);
+ List<CompletableFuture<Integer>> futures = Arrays.asList(
+ new CompletableFuture<>(),
+ new CompletableFuture<>(),
+ new CompletableFuture<>());
+ queue.scheduleDeferred("foo", __ -> OptionalLong.of(2L),
+ new FutureEvent<>(futures.get(0), () -> count.getAndIncrement()));
+ queue.append(new FutureEvent<>(futures.get(1), () -> count.getAndAdd(1)));
+ assertEquals(Integer.valueOf(0), futures.get(1).get());
+ time.sleep(1);
+ queue.append(new FutureEvent<>(futures.get(2), () -> count.getAndAdd(1)));
+ assertEquals(Integer.valueOf(1), futures.get(0).get());
+ assertEquals(Integer.valueOf(2), futures.get(2).get());
+ }
}
@Test
public void testShutdownBeforeDeferred() throws Exception {
- KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, new LogContext(),
- "testShutdownBeforeDeferred");
- final AtomicInteger count = new AtomicInteger(0);
- CompletableFuture<Integer> future = new CompletableFuture<>();
- queue.scheduleDeferred("myDeferred",
- __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + HOURS.toNanos(1)),
- new FutureEvent<>(future, () -> count.getAndAdd(1)));
- queue.beginShutdown("testShutdownBeforeDeferred");
- assertEquals(RejectedExecutionException.class, assertThrows(ExecutionException.class, () -> future.get()).getCause().getClass());
- assertEquals(0, count.get());
- queue.close();
+ try (KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext, "testShutdownBeforeDeferred")) {
+ final AtomicInteger count = new AtomicInteger(0);
+ CompletableFuture<Integer> future = new CompletableFuture<>();
+ queue.scheduleDeferred("myDeferred",
+ __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + HOURS.toNanos(1)),
+ new FutureEvent<>(future, () -> count.getAndAdd(1)));
+ queue.beginShutdown("testShutdownBeforeDeferred");
+ assertEquals(RejectedExecutionException.class, assertThrows(ExecutionException.class, () -> future.get()).getCause().getClass());
+ assertEquals(0, count.get());
+ }
}
@Test
public void testRejectedExecutionException() throws Exception {
- KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, new LogContext(),
+ KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext,
"testRejectedExecutionException");
queue.close();
CompletableFuture<Void> future = new CompletableFuture<>();
@@ -246,30 +259,31 @@ public class KafkaEventQueueTest {
@Test
public void testSize() throws Exception {
- KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, new LogContext(),
- "testEmpty");
- assertTrue(queue.isEmpty());
- CompletableFuture<Void> future = new CompletableFuture<>();
- queue.append(() -> future.get());
- assertFalse(queue.isEmpty());
- assertEquals(1, queue.size());
- queue.append(() -> future.get());
- assertEquals(2, queue.size());
- future.complete(null);
- TestUtils.waitForCondition(() -> queue.isEmpty(), "Failed to see the queue become empty.");
- queue.scheduleDeferred("later",
- __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + HOURS.toNanos(1)),
- () -> { });
- assertFalse(queue.isEmpty());
- queue.scheduleDeferred("soon",
- __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + TimeUnit.MILLISECONDS.toNanos(1)),
- () -> { });
- assertFalse(queue.isEmpty());
- queue.cancelDeferred("later");
- queue.cancelDeferred("soon");
- TestUtils.waitForCondition(() -> queue.isEmpty(), "Failed to see the queue become empty.");
- queue.close();
- assertTrue(queue.isEmpty());
+ try (KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext, "testEmpty")) {
+ assertTrue(queue.isEmpty());
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ queue.append(() -> future.get());
+ assertFalse(queue.isEmpty());
+ assertEquals(1, queue.size());
+ queue.append(() -> future.get());
+ assertEquals(2, queue.size());
+ future.complete(null);
+ TestUtils.waitForCondition(() -> queue.isEmpty(), "Failed to see the queue become empty.");
+ queue.scheduleDeferred("later",
+ __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + HOURS.toNanos(1)),
+ () -> {
+ });
+ assertFalse(queue.isEmpty());
+ queue.scheduleDeferred("soon",
+ __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + TimeUnit.MILLISECONDS.toNanos(1)),
+ () -> {
+ });
+ assertFalse(queue.isEmpty());
+ queue.cancelDeferred("later");
+ queue.cancelDeferred("soon");
+ TestUtils.waitForCondition(() -> queue.isEmpty(), "Failed to see the queue become empty.");
+ assertTrue(queue.isEmpty());
+ }
}
/**
@@ -277,32 +291,31 @@ public class KafkaEventQueueTest {
*/
@Test
public void testHandleExceptionThrowingAnException() throws Exception {
- KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, new LogContext(),
- "testHandleExceptionThrowingAnException");
- CompletableFuture<Void> initialFuture = new CompletableFuture<>();
- queue.append(() -> initialFuture.get());
- AtomicInteger counter = new AtomicInteger(0);
- queue.append(new EventQueue.Event() {
- @Override
- public void run() {
- counter.incrementAndGet();
- throw new IllegalStateException("First exception");
- }
-
- @Override
- public void handleException(Throwable e) {
- if (e instanceof IllegalStateException) {
+ try (KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext, "testHandleExceptionThrowingAnException")) {
+ CompletableFuture<Void> initialFuture = new CompletableFuture<>();
+ queue.append(() -> initialFuture.get());
+ AtomicInteger counter = new AtomicInteger(0);
+ queue.append(new EventQueue.Event() {
+ @Override
+ public void run() {
counter.incrementAndGet();
- throw new RuntimeException("Second exception");
+ throw new IllegalStateException("First exception");
}
- }
- });
- queue.append(() -> counter.incrementAndGet());
- assertEquals(3, queue.size());
- initialFuture.complete(null);
- TestUtils.waitForCondition(() -> counter.get() == 3,
- "Failed to see all events execute as planned.");
- queue.close();
+
+ @Override
+ public void handleException(Throwable e) {
+ if (e instanceof IllegalStateException) {
+ counter.incrementAndGet();
+ throw new RuntimeException("Second exception");
+ }
+ }
+ });
+ queue.append(() -> counter.incrementAndGet());
+ assertEquals(3, queue.size());
+ initialFuture.complete(null);
+ TestUtils.waitForCondition(() -> counter.get() == 3,
+ "Failed to see all events execute as planned.");
+ }
}
private static class InterruptibleEvent implements EventQueue.Event {
@@ -340,21 +353,20 @@ public class KafkaEventQueueTest {
@Test
public void testInterruptedExceptionHandling() throws Exception {
- KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, new LogContext(),
- "testInterruptedExceptionHandling");
- CompletableFuture<Thread> queueThread = new CompletableFuture<>();
- AtomicInteger numCallsToRun = new AtomicInteger(0);
- AtomicInteger numInterruptedExceptionsSeen = new AtomicInteger(0);
- queue.append(new InterruptibleEvent(queueThread, numCallsToRun, numInterruptedExceptionsSeen));
- queue.append(new InterruptibleEvent(queueThread, numCallsToRun, numInterruptedExceptionsSeen));
- queue.append(new InterruptibleEvent(queueThread, numCallsToRun, numInterruptedExceptionsSeen));
- queue.append(new InterruptibleEvent(queueThread, numCallsToRun, numInterruptedExceptionsSeen));
- queueThread.get().interrupt();
- TestUtils.retryOnExceptionWithTimeout(30000,
- () -> assertEquals(1, numCallsToRun.get()));
- TestUtils.retryOnExceptionWithTimeout(30000,
- () -> assertEquals(3, numInterruptedExceptionsSeen.get()));
- queue.close();
+ try (KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext, "testInterruptedExceptionHandling")) {
+ CompletableFuture<Thread> queueThread = new CompletableFuture<>();
+ AtomicInteger numCallsToRun = new AtomicInteger(0);
+ AtomicInteger numInterruptedExceptionsSeen = new AtomicInteger(0);
+ queue.append(new InterruptibleEvent(queueThread, numCallsToRun, numInterruptedExceptionsSeen));
+ queue.append(new InterruptibleEvent(queueThread, numCallsToRun, numInterruptedExceptionsSeen));
+ queue.append(new InterruptibleEvent(queueThread, numCallsToRun, numInterruptedExceptionsSeen));
+ queue.append(new InterruptibleEvent(queueThread, numCallsToRun, numInterruptedExceptionsSeen));
+ queueThread.get().interrupt();
+ TestUtils.retryOnExceptionWithTimeout(30000,
+ () -> assertEquals(1, numCallsToRun.get()));
+ TestUtils.retryOnExceptionWithTimeout(30000,
+ () -> assertEquals(3, numInterruptedExceptionsSeen.get()));
+ }
}
static class ExceptionTrapperEvent implements EventQueue.Event {
@@ -374,42 +386,41 @@ public class KafkaEventQueueTest {
@Test
public void testInterruptedWithEmptyQueue() throws Exception {
CompletableFuture<Void> cleanupFuture = new CompletableFuture<>();
- KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, new LogContext(),
- "testInterruptedWithEmptyQueue", () -> cleanupFuture.complete(null));
- CompletableFuture<Thread> queueThread = new CompletableFuture<>();
- queue.append(() -> queueThread.complete(Thread.currentThread()));
- TestUtils.retryOnExceptionWithTimeout(30000, () -> assertEquals(0, queue.size()));
- queueThread.get().interrupt();
- cleanupFuture.get();
- ExceptionTrapperEvent ieTrapper = new ExceptionTrapperEvent();
- queue.append(ieTrapper);
- assertEquals(InterruptedException.class, ieTrapper.exception.get().getClass());
- queue.close();
- ExceptionTrapperEvent reTrapper = new ExceptionTrapperEvent();
- queue.append(reTrapper);
- assertEquals(RejectedExecutionException.class, reTrapper.exception.get().getClass());
+ try (KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext, "testInterruptedWithEmptyQueue", () -> cleanupFuture.complete(null))) {
+ CompletableFuture<Thread> queueThread = new CompletableFuture<>();
+ queue.append(() -> queueThread.complete(Thread.currentThread()));
+ TestUtils.retryOnExceptionWithTimeout(30000, () -> assertEquals(0, queue.size()));
+ queueThread.get().interrupt();
+ cleanupFuture.get();
+ ExceptionTrapperEvent ieTrapper = new ExceptionTrapperEvent();
+ queue.append(ieTrapper);
+ assertEquals(InterruptedException.class, ieTrapper.exception.get().getClass());
+ queue.close();
+ ExceptionTrapperEvent reTrapper = new ExceptionTrapperEvent();
+ queue.append(reTrapper);
+ assertEquals(RejectedExecutionException.class, reTrapper.exception.get().getClass());
+ }
}
@Test
public void testInterruptedWithDeferredEvents() throws Exception {
CompletableFuture<Void> cleanupFuture = new CompletableFuture<>();
- KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, new LogContext(),
- "testInterruptedWithDeferredEvents", () -> cleanupFuture.complete(null));
- CompletableFuture<Thread> queueThread = new CompletableFuture<>();
- queue.append(() -> queueThread.complete(Thread.currentThread()));
- ExceptionTrapperEvent ieTrapper1 = new ExceptionTrapperEvent();
- ExceptionTrapperEvent ieTrapper2 = new ExceptionTrapperEvent();
- queue.scheduleDeferred("ie2",
- __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + HOURS.toNanos(2)),
- ieTrapper2);
- queue.scheduleDeferred("ie1",
- __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + HOURS.toNanos(1)),
- ieTrapper1);
- TestUtils.retryOnExceptionWithTimeout(30000, () -> assertEquals(2, queue.size()));
- queueThread.get().interrupt();
- cleanupFuture.get();
- assertEquals(InterruptedException.class, ieTrapper1.exception.get().getClass());
- assertEquals(InterruptedException.class, ieTrapper2.exception.get().getClass());
- queue.close();
+ try (KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext, "testInterruptedWithDeferredEvents", () -> cleanupFuture.complete(null))) {
+ CompletableFuture<Thread> queueThread = new CompletableFuture<>();
+ queue.append(() -> queueThread.complete(Thread.currentThread()));
+ ExceptionTrapperEvent ieTrapper1 = new ExceptionTrapperEvent();
+ ExceptionTrapperEvent ieTrapper2 = new ExceptionTrapperEvent();
+ queue.scheduleDeferred("ie2",
+ __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + HOURS.toNanos(2)),
+ ieTrapper2);
+ queue.scheduleDeferred("ie1",
+ __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + HOURS.toNanos(1)),
+ ieTrapper1);
+ TestUtils.retryOnExceptionWithTimeout(30000, () -> assertEquals(2, queue.size()));
+ queueThread.get().interrupt();
+ cleanupFuture.get();
+ assertEquals(InterruptedException.class, ieTrapper1.exception.get().getClass());
+ assertEquals(InterruptedException.class, ieTrapper2.exception.get().getClass());
+ }
}
}