You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/09/14 06:29:28 UTC
(pulsar) branch master updated: [improve][client] Coalesce message listener drain scheduling (#26531)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new a6278f292533 [improve][client] Coalesce message listener drain scheduling (#26531)
a6278f292533 is described below
commit a6278f292533259274a0246cc2a08c8f463ae8e0
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Sep 14 09:29:22 2026 +0300
[improve][client] Coalesce message listener drain scheduling (#26531)
---
.../impl/ListenerTaskSchedulingBenchmark.java | 102 +++++++++
.../apache/pulsar/client/impl/ConsumerBase.java | 49 +++--
.../pulsar/client/impl/ListenerTaskScheduler.java | 73 +++++++
.../impl/ConsumerListenerSchedulingTest.java | 170 ++++++++++++++
.../client/impl/ListenerTaskSchedulerTest.java | 243 +++++++++++++++++++++
5 files changed, 619 insertions(+), 18 deletions(-)
diff --git a/microbench/src/main/java/org/apache/pulsar/client/impl/ListenerTaskSchedulingBenchmark.java b/microbench/src/main/java/org/apache/pulsar/client/impl/ListenerTaskSchedulingBenchmark.java
new file mode 100644
index 000000000000..282e631f4573
--- /dev/null
+++ b/microbench/src/main/java/org/apache/pulsar/client/impl/ListenerTaskSchedulingBenchmark.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import java.util.Queue;
+import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+/** Models queued arrival/drain scheduling, excluding message decoding, listener calls and OS thread handoff. */
+@State(Scope.Thread)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Fork(2)
+public class ListenerTaskSchedulingBenchmark {
+ @Param({"1", "16", "1024"})
+ public int notifications;
+
+ @Param({"fresh", "reused", "coalesced"})
+ public String scheduling;
+
+ private QueuedExecutor executor;
+ private ListenerTaskScheduler scheduler;
+ private Runnable arrival;
+ private Runnable trigger;
+ private int incoming;
+ private int received;
+
+ @Setup
+ public void setup() {
+ executor = new QueuedExecutor();
+ scheduler = new ListenerTaskScheduler(executor, this::drain);
+ arrival = () -> incoming++;
+ Runnable reusedDrain = this::drain;
+ trigger = switch (scheduling) {
+ case "fresh" -> () -> executor.execute(() -> drain());
+ case "reused" -> () -> executor.execute(reusedDrain);
+ case "coalesced" -> scheduler::trigger;
+ default -> throw new IllegalArgumentException(scheduling);
+ };
+ }
+
+ @Benchmark
+ public int notifyAndDrain() {
+ received = 0;
+ for (int i = 0; i < notifications; i++) {
+ executor.execute(arrival);
+ trigger.run();
+ }
+ Runnable task;
+ while ((task = executor.tasks.poll()) != null) {
+ task.run();
+ }
+ if (received != notifications || incoming != 0) {
+ throw new IllegalStateException("A notification was lost");
+ }
+ return received;
+ }
+
+ private void drain() {
+ received += incoming;
+ incoming = 0;
+ }
+
+ private static final class QueuedExecutor implements Executor {
+ private final Queue<Runnable> tasks = new LinkedBlockingQueue<>();
+
+ @Override
+ public void execute(Runnable command) {
+ tasks.add(command);
+ }
+ }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
index 1cde078be0c9..2c07a94fc698 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
@@ -92,6 +92,7 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
protected final MessageListenerExecutor messageListenerExecutor;
protected final ExecutorService externalPinnedExecutor;
protected final ExecutorService internalPinnedExecutor;
+ private final ListenerTaskScheduler listenerTaskScheduler;
protected final UnAckedMessageTracker unAckedMessageTracker;
final GrowableArrayBlockingQueue<Message<T>> incomingMessages;
protected Map<MessageIdAdv, MessageIdImpl[]> unAckedChunkedMessageIdSequenceMap = new ConcurrentHashMap<>();
@@ -164,6 +165,8 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
: conf.getMessageListenerExecutor();
this.externalPinnedExecutor = executorProvider.getExecutor();
this.internalPinnedExecutor = client.getInternalExecutorService();
+ this.listenerTaskScheduler = listener == null
+ ? null : new ListenerTaskScheduler(internalPinnedExecutor, this::drainListener);
this.pendingReceives = Queues.newConcurrentLinkedQueue();
this.pendingBatchReceives = Queues.newConcurrentLinkedQueue();
this.schema = schema;
@@ -1221,26 +1224,36 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
// The messages are added into the receiver queue by the internal pinned executor,
// so need to use internal pinned executor to avoid race condition which message
// might be added into the receiver queue but not able to read here.
- internalPinnedExecutor.execute(() -> {
- try {
- Message<T> msg;
- do {
- msg = internalReceive(0, TimeUnit.MILLISECONDS);
- if (msg != null) {
- // Trigger the notification on the message listener in a separate thread to avoid blocking the
- // internal pinned executor thread while the message processing happens
- final Message<T> finalMsg = msg;
- MESSAGE_LISTENER_QUEUE_SIZE_UPDATER.incrementAndGet(this);
+ listenerTaskScheduler.trigger();
+ }
+
+ private void drainListener() {
+ try {
+ Message<T> msg;
+ do {
+ msg = internalReceive(0, TimeUnit.MILLISECONDS);
+ if (msg != null) {
+ // Trigger the notification on the message listener in a separate thread to avoid blocking the
+ // internal pinned executor thread while the message processing happens
+ final Message<T> finalMsg = msg;
+ MESSAGE_LISTENER_QUEUE_SIZE_UPDATER.incrementAndGet(this);
+ try {
messageListenerExecutor.execute(msg, () -> callMessageListener(finalMsg));
- } else {
- log.debug("Message has been cleared from the queue");
+ } catch (RuntimeException | Error error) {
+ // A failed submission has dequeued a message, but may leave more messages to drain.
+ // Preserve a later turn even when earlier failures consumed the coalesced notification.
+ // Retrying here makes progress; retrying a failed dequeue could loop indefinitely.
+ listenerTaskScheduler.trigger();
+ throw error;
}
- } while (msg != null);
- } catch (PulsarClientException e) {
- log.warn().exception(e)
- .log("Failed to dequeue the message for listener");
- }
- });
+ } else {
+ log.debug("Message has been cleared from the queue");
+ }
+ } while (msg != null);
+ } catch (PulsarClientException e) {
+ log.warn().exception(e)
+ .log("Failed to dequeue the message for listener");
+ }
}
private void executeMessageListener(Message<?> message, Runnable runnable) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ListenerTaskScheduler.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ListenerTaskScheduler.java
new file mode 100644
index 000000000000..821ae1693621
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ListenerTaskScheduler.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
+/**
+ * Coalesces listener-drain requests on an asynchronous, serial FIFO executor.
+ * An additional trigger requires a later turn: its arrival task may be queued behind the current drain.
+ * Arrival tasks must be enqueued before their corresponding triggers. Rejected submissions reset the state
+ * for a later notification to retry; delivery is not guaranteed while the executor rejects work.
+ */
+final class ListenerTaskScheduler implements Runnable {
+ private static final int IDLE = 0;
+ private static final int SCHEDULED = 1;
+ private static final int TRIGGERED_AGAIN = 2;
+ private static final AtomicIntegerFieldUpdater<ListenerTaskScheduler> STATE_UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(ListenerTaskScheduler.class, "state");
+
+ private final Executor executor;
+ private final Runnable drain;
+ private volatile int state;
+
+ ListenerTaskScheduler(Executor executor, Runnable drain) {
+ this.executor = executor;
+ this.drain = drain;
+ }
+
+ void trigger() {
+ int previous = STATE_UPDATER.getAndUpdate(this,
+ current -> current == IDLE ? SCHEDULED : TRIGGERED_AGAIN);
+ if (previous == IDLE) {
+ try {
+ executor.execute(this);
+ } catch (RejectedExecutionException error) {
+ // A later notification can retry if the executor becomes available again.
+ STATE_UPDATER.set(this, IDLE);
+ throw error;
+ }
+ }
+ }
+
+ @Override
+ public void run() {
+ try {
+ drain.run();
+ } finally {
+ if (STATE_UPDATER.getAndSet(this, IDLE) == TRIGGERED_AGAIN) {
+ // Enqueue after pending arrival tasks, rather than draining again in this turn.
+ // A concurrent trigger after the reset either schedules this turn or marks it for a later turn.
+ trigger();
+ }
+ }
+ }
+}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerListenerSchedulingTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerListenerSchedulingTest.java
new file mode 100644
index 000000000000..19dedd85a480
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerListenerSchedulingTest.java
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.testng.annotations.Test;
+
+public class ConsumerListenerSchedulingTest {
+ @Test
+ public void consecutiveRejectionsDoNotStrandCoalescedMessages() {
+ Fixture fixture = new Fixture(2);
+ Message<byte[]> first = newMessage();
+ Message<byte[]> second = newMessage();
+ Message<byte[]> third = newMessage();
+ fixture.addAndTrigger(first);
+ fixture.addAndTrigger(second);
+ fixture.addAndTrigger(third);
+
+ assertEquals(fixture.tasks.size(), 1);
+ expectThrows(RejectedExecutionException.class, fixture::runNext);
+ expectThrows(RejectedExecutionException.class, fixture::runNext);
+ fixture.runRemaining();
+
+ assertEquals(fixture.attempted, List.of(first, second, third));
+ assertEquals(fixture.accepted, List.of(third));
+ assertTrue(fixture.consumer.incomingMessages.isEmpty());
+ }
+
+ @Test
+ public void rejectingEveryMessageFinishesAndAllowsLaterNotifications() {
+ Fixture fixture = new Fixture(3);
+ for (int i = 0; i < 3; i++) {
+ fixture.addAndTrigger(newMessage());
+ }
+ for (int i = 0; i < 3; i++) {
+ expectThrows(RejectedExecutionException.class, fixture::runNext);
+ }
+ fixture.runRemaining();
+ assertEquals(fixture.attempted.size(), 3);
+ assertTrue(fixture.accepted.isEmpty());
+ assertTrue(fixture.consumer.incomingMessages.isEmpty());
+
+ Message<byte[]> next = newMessage();
+ fixture.addAndTrigger(next);
+ fixture.runRemaining();
+ assertEquals(fixture.accepted, List.of(next));
+ }
+
+ @Test
+ public void rejectionRetryRunsAfterPendingArrivals() {
+ Fixture fixture = new Fixture(2);
+ Message<byte[]> first = newMessage();
+ Message<byte[]> second = newMessage();
+ Message<byte[]> third = newMessage();
+ for (Message<byte[]> message : List.of(first, second, third)) {
+ fixture.tasks.add(() -> fixture.consumer.incomingMessages.add(message));
+ fixture.consumer.tryTriggerListener();
+ }
+
+ fixture.runNext(); // First arrival.
+ expectThrows(RejectedExecutionException.class, fixture::runNext);
+ assertEquals(fixture.attempted, List.of(first));
+ assertTrue(fixture.consumer.incomingMessages.isEmpty());
+ fixture.runNext(); // Second arrival, ahead of the retry.
+ fixture.runNext(); // Third arrival, ahead of the retry.
+ assertEquals(fixture.consumer.incomingMessages.size(), 2);
+ expectThrows(RejectedExecutionException.class, fixture::runNext);
+ fixture.runRemaining();
+
+ assertEquals(fixture.attempted, List.of(first, second, third));
+ assertEquals(fixture.accepted, List.of(third));
+ assertTrue(fixture.consumer.incomingMessages.isEmpty());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Message<byte[]> newMessage() {
+ return mock(Message.class);
+ }
+
+ private static final class Fixture {
+ private final Queue<Runnable> tasks = new ArrayDeque<>();
+ private final List<Message<?>> attempted = new ArrayList<>();
+ private final List<Message<?>> accepted = new ArrayList<>();
+ private final ConsumerImpl<byte[]> consumer;
+
+ private Fixture(int rejectedSubmissions) {
+ ExecutorService executor = mock(ExecutorService.class);
+ doAnswer(invocation -> {
+ tasks.add(invocation.getArgument(0));
+ return null;
+ }).when(executor).execute(any(Runnable.class));
+ ExecutorProvider executorProvider = mock(ExecutorProvider.class);
+ when(executorProvider.getExecutor()).thenReturn(executor);
+ PulsarClientImpl client = ClientTestFixtures.createPulsarClientMock(executorProvider, executor);
+ client.getConfiguration().setStatsIntervalSeconds(0);
+ ConsumerConfigurationData<byte[]> conf = new ConsumerConfigurationData<>();
+ conf.setSubscriptionName("test-sub");
+ conf.setMessageListener((ignored, message) -> { });
+ conf.setMessageListenerExecutor((message, runnable) -> {
+ attempted.add(message);
+ if (attempted.size() <= rejectedSubmissions) {
+ throw new RejectedExecutionException("listener submission rejected");
+ }
+ accepted.add(message);
+ });
+ consumer = new ConsumerImpl<>(client, "non-persistent://tenant/ns/listener-scheduling", conf,
+ executorProvider, -1, false, false, new CompletableFuture<>(), null, 0, Schema.BYTES,
+ null, true) {
+ @Override
+ protected Message<byte[]> internalReceive(long timeout, TimeUnit unit) {
+ // Keep the real listener drain and scheduler, without message decoding or broker state.
+ return incomingMessages.poll();
+ }
+ };
+ consumer.setState(HandlerState.State.Ready);
+ assertTrue(tasks.isEmpty());
+ }
+
+ private void addAndTrigger(Message<byte[]> message) {
+ consumer.incomingMessages.add(message);
+ consumer.tryTriggerListener();
+ }
+
+ private void runNext() {
+ tasks.remove().run();
+ }
+
+ private void runRemaining() {
+ int runs = 0;
+ while (!tasks.isEmpty()) {
+ assertTrue(++runs <= 10, "Listener drain must not retry indefinitely");
+ runNext();
+ }
+ }
+ }
+}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ListenerTaskSchedulerTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ListenerTaskSchedulerTest.java
new file mode 100644
index 000000000000..c11b1af15d67
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ListenerTaskSchedulerTest.java
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.testng.annotations.Test;
+
+public class ListenerTaskSchedulerTest {
+ @Test
+ public void drainsArrivalQueuedBehindAnAlreadyScheduledDrain() {
+ QueuedExecutor executor = new QueuedExecutor();
+ Queue<Integer> arrivals = new ArrayDeque<>();
+ List<Integer> received = new ArrayList<>();
+ AtomicInteger runs = new AtomicInteger();
+ ListenerTaskScheduler scheduler = new ListenerTaskScheduler(executor, () -> {
+ runs.incrementAndGet();
+ Integer arrival;
+ while ((arrival = arrivals.poll()) != null) {
+ received.add(arrival);
+ }
+ });
+ executor.execute(() -> arrivals.add(1));
+ scheduler.trigger();
+ executor.execute(() -> arrivals.add(2));
+ scheduler.trigger();
+ assertEquals(executor.tasks.size(), 3);
+ executor.runNext();
+ executor.runNext();
+ assertEquals(received, List.of(1));
+ executor.runAll();
+ assertEquals(received, List.of(1, 2));
+ assertEquals(runs.get(), 2);
+ }
+
+ @Test
+ public void coalescesABurstAndReturnsToIdle() {
+ QueuedExecutor executor = new QueuedExecutor();
+ AtomicInteger runs = new AtomicInteger();
+ ListenerTaskScheduler scheduler = new ListenerTaskScheduler(executor, runs::incrementAndGet);
+ for (int i = 0; i < 1000; i++) {
+ scheduler.trigger();
+ }
+ assertEquals(executor.tasks.size(), 1);
+ executor.runAll();
+ assertEquals(runs.get(), 2);
+ scheduler.trigger();
+ executor.runAll();
+ assertEquals(runs.get(), 3);
+ }
+
+ @Test
+ public void triggerDuringDrainSchedulesAnotherTurn() {
+ QueuedExecutor executor = new QueuedExecutor();
+ AtomicInteger runs = new AtomicInteger();
+ AtomicReference<ListenerTaskScheduler> reference = new AtomicReference<>();
+ ListenerTaskScheduler scheduler = new ListenerTaskScheduler(executor, () -> {
+ if (runs.incrementAndGet() == 1) {
+ reference.get().trigger();
+ }
+ });
+ reference.set(scheduler);
+ scheduler.trigger();
+ executor.runAll();
+ assertEquals(runs.get(), 2);
+ }
+
+ @Test
+ public void drainFailurePreservesLaterTriggers() {
+ QueuedExecutor executor = new QueuedExecutor();
+ AtomicInteger runs = new AtomicInteger();
+ ListenerTaskScheduler scheduler = new ListenerTaskScheduler(executor, () -> {
+ if (runs.incrementAndGet() == 1) {
+ throw new IllegalStateException("drain failed");
+ }
+ });
+ scheduler.trigger();
+ scheduler.trigger();
+ expectThrows(IllegalStateException.class, executor::runNext);
+ executor.runAll();
+ assertEquals(runs.get(), 2);
+ }
+
+ @Test
+ public void rejectedSubmissionAllowsLaterRetry() {
+ QueuedExecutor executor = new QueuedExecutor();
+ AtomicInteger submissions = new AtomicInteger();
+ AtomicInteger runs = new AtomicInteger();
+ ListenerTaskScheduler scheduler = new ListenerTaskScheduler(task -> {
+ if (submissions.incrementAndGet() == 1) {
+ throw new RejectedExecutionException("first submission rejected");
+ }
+ executor.execute(task);
+ }, runs::incrementAndGet);
+ expectThrows(RejectedExecutionException.class, scheduler::trigger);
+ scheduler.trigger();
+ executor.runAll();
+ assertEquals(runs.get(), 1);
+ }
+
+ @Test
+ public void rejectedFollowUpAllowsLaterRetry() {
+ QueuedExecutor executor = new QueuedExecutor();
+ AtomicInteger submissions = new AtomicInteger();
+ AtomicInteger runs = new AtomicInteger();
+ ListenerTaskScheduler scheduler = new ListenerTaskScheduler(task -> {
+ if (submissions.incrementAndGet() == 2) {
+ throw new RejectedExecutionException("follow-up submission rejected");
+ }
+ executor.execute(task);
+ }, runs::incrementAndGet);
+ scheduler.trigger();
+ scheduler.trigger();
+ expectThrows(RejectedExecutionException.class, executor::runNext);
+ assertEquals(runs.get(), 1);
+ assertTrue(executor.tasks.isEmpty());
+ scheduler.trigger();
+ executor.runAll();
+ assertEquals(runs.get(), 2);
+ }
+
+ @Test
+ public void drainFailureReturnsToIdle() {
+ QueuedExecutor executor = new QueuedExecutor();
+ AtomicInteger runs = new AtomicInteger();
+ ListenerTaskScheduler scheduler = new ListenerTaskScheduler(executor, () -> {
+ if (runs.incrementAndGet() == 1) {
+ throw new IllegalStateException("drain failed");
+ }
+ });
+ scheduler.trigger();
+ expectThrows(IllegalStateException.class, executor::runNext);
+ assertTrue(executor.tasks.isEmpty());
+ scheduler.trigger();
+ executor.runAll();
+ assertEquals(runs.get(), 2);
+ }
+
+ @Test(invocationCount = 5, timeOut = 30000)
+ public void concurrentNotificationsDoNotLoseArrivals() throws Exception {
+ int producerCount = 4;
+ int perProducer = 5000;
+ int total = producerCount * perProducer;
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ ExecutorService producers = Executors.newFixedThreadPool(producerCount);
+ Queue<Integer> arrivals = new ArrayDeque<>();
+ BitSet received = new BitSet(total);
+ AtomicInteger duplicates = new AtomicInteger();
+ CountDownLatch complete = new CountDownLatch(total);
+ CountDownLatch start = new CountDownLatch(1);
+ ListenerTaskScheduler scheduler = new ListenerTaskScheduler(executor, () -> {
+ Integer id;
+ while ((id = arrivals.poll()) != null) {
+ if (received.get(id)) {
+ duplicates.incrementAndGet();
+ }
+ received.set(id);
+ complete.countDown();
+ }
+ });
+ try {
+ List<Future<?>> jobs = new ArrayList<>();
+ for (int producer = 0; producer < producerCount; producer++) {
+ int first = producer * perProducer;
+ jobs.add(producers.submit(() -> {
+ start.await();
+ for (int i = 0; i < perProducer; i++) {
+ int id = first + i;
+ executor.execute(() -> arrivals.add(id));
+ scheduler.trigger();
+ }
+ return null;
+ }));
+ }
+ start.countDown();
+ for (Future<?> job : jobs) {
+ job.get(10, TimeUnit.SECONDS);
+ }
+ assertTrue(complete.await(10, TimeUnit.SECONDS), "An arrival was left without a drain");
+ executor.submit(() -> {
+ assertEquals(received.cardinality(), total);
+ assertEquals(duplicates.get(), 0);
+ assertTrue(arrivals.isEmpty());
+ }).get(10, TimeUnit.SECONDS);
+ } finally {
+ start.countDown();
+ producers.shutdownNow();
+ executor.shutdownNow();
+ assertTrue(producers.awaitTermination(5, TimeUnit.SECONDS));
+ assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+ private static final class QueuedExecutor implements Executor {
+ private final Queue<Runnable> tasks = new ArrayDeque<>();
+
+ @Override
+ public void execute(Runnable command) {
+ tasks.add(command);
+ }
+
+ void runNext() {
+ tasks.remove().run();
+ }
+
+ void runAll() {
+ while (!tasks.isEmpty()) {
+ runNext();
+ }
+ }
+ }
+}