You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2026/10/17 13:08:10 UTC
(fluss) branch main updated: [Server] improve CI test time for netty (#4645)
This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 6b43eaba5 [Server] improve CI test time for netty (#4645)
6b43eaba5 is described below
commit 6b43eaba5886fac3ac58b343cdf0b2f2c0222fdf
Author: Giannis Polyzos <ip...@gmail.com>
AuthorDate: Sat Oct 17 14:08:05 2026 +0100
[Server] improve CI test time for netty (#4645)
---
.../flink/FlinkUnionReadPrimaryKeyTableITCase.java | 8 +-
.../org/apache/fluss/rpc/netty/NettyUtils.java | 12 ++-
.../apache/fluss/rpc/netty/client/NettyClient.java | 46 ++++++----
.../apache/fluss/rpc/netty/server/NettyServer.java | 20 +++--
.../fluss/rpc/netty/server/RequestChannel.java | 70 +++++++++------
.../fluss/rpc/netty/client/NettyClientTest.java | 26 ++++++
.../fluss/rpc/netty/server/NettyServerTest.java | 99 ++++++++++++++++++++++
.../fluss/rpc/protocol/RequestChannelTest.java | 43 ++++++++++
8 files changed, 275 insertions(+), 49 deletions(-)
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java
index fbeab1777..b4ce39c96 100644
--- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java
+++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java
@@ -1127,6 +1127,7 @@ class FlinkUnionReadPrimaryKeyTableITCase extends FlinkUnionReadTestBase {
// 3. Start tiering job
JobClient jobClient = buildTieringJob(execEnv);
+ boolean tieringCancelled = false;
try {
// 4. Wait for data to snapshot to Paimon
assertReplicaStatus(tableBucket, 2);
@@ -1164,7 +1165,9 @@ class FlinkUnionReadPrimaryKeyTableITCase extends FlinkUnionReadTestBase {
// 9. Add Column "c4" (Schema V3)
// Verify union read reconciles tiered data (V1) with a fluss log
// containing multiple schema versions (V2 and V3).
+ // The per-job MiniCluster shuts down with the job, so cancel exactly once.
jobClient.cancel().get();
+ tieringCancelled = true;
addColumnChanges =
Collections.singletonList(
TableChange.addColumn(
@@ -1192,7 +1195,10 @@ class FlinkUnionReadPrimaryKeyTableITCase extends FlinkUnionReadTestBase {
"+I[2, v2_updated_again, 20, 30]",
"+I[3, v3_update, 30, 40]");
} finally {
- jobClient.cancel().get();
+ // only cancel here if the test failed before the intended cancel above
+ if (!tieringCancelled) {
+ jobClient.cancel().get();
+ }
}
}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/NettyUtils.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/NettyUtils.java
index b6418e6dc..1cba984d2 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/NettyUtils.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/NettyUtils.java
@@ -32,6 +32,7 @@ import org.apache.fluss.shaded.netty4.io.netty.channel.unix.Errors;
import org.apache.fluss.shaded.netty4.io.netty.util.concurrent.DefaultThreadFactory;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
/** Utils of netty. */
public class NettyUtils {
@@ -70,10 +71,19 @@ public class NettyUtils {
}
}
+ /**
+ * Shuts down the given event loop group without a quiet period, closing all of its channels.
+ *
+ * <p>Netty's default 2 second quiet period keeps accepting tasks that other threads submit
+ * after the shutdown started, and delays every shutdown by at least 2 seconds. Without it, the
+ * group terminates as soon as its pending tasks have run, and later submissions are rejected
+ * with a {@link java.util.concurrent.RejectedExecutionException}. Callers must therefore only
+ * shut down a group once nothing submits work to it anymore, or tolerate the rejection.
+ */
public static CompletableFuture<Void> shutdownGroup(EventLoopGroup group) {
CompletableFuture<Void> shutdownFuture = new CompletableFuture<>();
if (group != null) {
- group.shutdownGracefully()
+ group.shutdownGracefully(0, 10, TimeUnit.SECONDS)
.addListener(
finished -> {
if (finished.isSuccess()) {
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java
index 50e714d02..f1c1bb0d1 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java
@@ -171,28 +171,44 @@ public final class NettyClient implements RpcClient {
shutdownFutures.add(conn.getValue().close());
}
}
- shutdownFutures.add(NettyUtils.shutdownGroup(eventGroup));
- CompletableFuture.allOf(shutdownFutures.toArray(new CompletableFuture<?>[0]))
- .get(10, TimeUnit.SECONDS);
+ try {
+ CompletableFuture.allOf(shutdownFutures.toArray(new CompletableFuture<?>[0]))
+ .get(10, TimeUnit.SECONDS);
+ } finally {
+ // Shut down the event loop only after the connections are closed, as it
+ // terminates without a quiet period. It is shut down even if closing a
+ // connection failed or timed out.
+ NettyUtils.shutdownGroup(eventGroup).get(10, TimeUnit.SECONDS);
+ }
LOG.info("Netty client was shutdown successfully.");
} catch (Exception e) {
LOG.warn("Netty client shutdown failed: ", e);
}
}
- private ServerConnection getOrCreateConnection(ServerNode node) {
+ @VisibleForTesting
+ ServerConnection getOrCreateConnection(ServerNode node) {
String serverId = node.uid();
- return connections.computeIfAbsent(
- serverId,
- ignored -> {
- LOG.debug("Creating connection to server {}.", node);
- return new ServerConnection(
- bootstrap,
- node,
- clientMetricGroup,
- authenticatorSupplier.get(),
- (con, ignore) -> connections.remove(serverId, con));
- });
+ ServerConnection connection =
+ connections.computeIfAbsent(
+ serverId,
+ ignored -> {
+ LOG.debug("Creating connection to server {}.", node);
+ return new ServerConnection(
+ bootstrap,
+ node,
+ clientMetricGroup,
+ authenticatorSupplier.get(),
+ (con, ignore) -> connections.remove(serverId, con));
+ });
+ // close() may have run before the connection was added and missed it. Its event loop is
+ // then terminated, so the connection would never connect and its requests would never
+ // complete. Close it here, so they fail instead. isClosed is volatile: either this check
+ // sees the client closed, or close() sees the connection.
+ if (isClosed) {
+ connection.close();
+ }
+ return connection;
}
@VisibleForTesting
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java
index df2256985..fa0174cb4 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java
@@ -304,24 +304,28 @@ public final class NettyServer implements RpcServer {
isRunning = false;
- CompletableFuture<Void> acceptorShutdownFuture = shutdownGroup(acceptorGroup);
- CompletableFuture<Void> selectorShutdownFuture = shutdownGroup(selectorGroup);
+ // Stop accepting new connections.
CompletableFuture<Void> channelShutdownFuture =
FutureUtils.completeAll(
bindChannels.stream()
.map(NettyUtils::shutdownChannel)
.collect(Collectors.toList()));
+ // Drain the requests that were already accepted.
CompletableFuture<Void> workerShutdownFuture;
if (workerPool != null) {
workerShutdownFuture = workerPool.closeAsync();
} else {
workerShutdownFuture = CompletableFuture.completedFuture(null);
}
-
- return CompletableFuture.allOf(
- acceptorShutdownFuture,
- selectorShutdownFuture,
- channelShutdownFuture,
- workerShutdownFuture);
+ // Shut down the event loops only after both of the above completed. The event loops
+ // terminate without a quiet period, so they must outlive the tasks still submitted to
+ // them: the bind channel close listeners, and the responses (and backpressure updates)
+ // of the requests drained by the worker pool. Shutting the event loops down closes the
+ // remaining client connections.
+ return FutureUtils.composeAfterwards(
+ CompletableFuture.allOf(channelShutdownFuture, workerShutdownFuture),
+ () ->
+ CompletableFuture.allOf(
+ shutdownGroup(acceptorGroup), shutdownGroup(selectorGroup)));
}
}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestChannel.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestChannel.java
index 8ee85d500..3b52bdce3 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestChannel.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestChannel.java
@@ -28,6 +28,7 @@ import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
@@ -224,18 +225,18 @@ public class RequestChannel {
for (Channel channel : associatedChannels) {
if (channel.isActive()) {
// Submit to the channel's EventLoop to ensure thread safety
- channel.eventLoop()
- .execute(
- () -> {
- if (channel.isActive() && channel.config().isAutoRead()) {
- channel.config().setAutoRead(false);
- LOG.warn(
- "Queue size ({}) reached backpressure threshold ({}), paused channel: {}",
- queueSize,
- backpressureThreshold,
- channel.remoteAddress());
- }
- });
+ executeOnEventLoop(
+ channel,
+ () -> {
+ if (channel.isActive() && channel.config().isAutoRead()) {
+ channel.config().setAutoRead(false);
+ LOG.warn(
+ "Queue size ({}) reached backpressure threshold ({}), paused channel: {}",
+ queueSize,
+ backpressureThreshold,
+ channel.remoteAddress());
+ }
+ });
}
}
}
@@ -244,19 +245,40 @@ public class RequestChannel {
for (Channel channel : associatedChannels) {
if (channel.isActive()) {
// Submit resume task to the channel's EventLoop to ensure thread safety
- channel.eventLoop()
- .execute(
- () -> {
- if (channel.isActive() && !channel.config().isAutoRead()) {
- channel.config().setAutoRead(true);
- LOG.info(
- "Queue size ({}) reached resume threshold ({}), resumed channel: {}",
- queueSize,
- resumeThreshold,
- channel.remoteAddress());
- }
- });
+ executeOnEventLoop(
+ channel,
+ () -> {
+ if (channel.isActive() && !channel.config().isAutoRead()) {
+ channel.config().setAutoRead(true);
+ LOG.info(
+ "Queue size ({}) reached resume threshold ({}), resumed channel: {}",
+ queueSize,
+ resumeThreshold,
+ channel.remoteAddress());
+ }
+ });
}
}
}
+
+ /**
+ * Executes the task on the channel's EventLoop. The EventLoop may already be shutting down
+ * (e.g. during server shutdown) even though the channel was still active when checked; in that
+ * case the task is skipped, because the channel is about to be closed anyway. Letting the
+ * rejection propagate would kill the calling request processor thread, and its shutdown future
+ * would never complete.
+ *
+ * <p>A skipped task leaves {@link #isBackpressureActive} out of sync with the channel's actual
+ * auto-read state. This is harmless, as the channel is closed together with its EventLoop.
+ */
+ private static void executeOnEventLoop(Channel channel, Runnable task) {
+ try {
+ channel.eventLoop().execute(task);
+ } catch (RejectedExecutionException e) {
+ LOG.debug(
+ "Skip backpressure task for channel {} because its event loop is shutting down.",
+ channel.remoteAddress(),
+ e);
+ }
+ }
}
diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/NettyClientTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/NettyClientTest.java
index 380158f27..4e5d334d7 100644
--- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/NettyClientTest.java
+++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/client/NettyClientTest.java
@@ -22,6 +22,7 @@ import org.apache.fluss.cluster.ServerNode;
import org.apache.fluss.cluster.ServerType;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.NetworkException;
import org.apache.fluss.metrics.groups.MetricGroup;
import org.apache.fluss.metrics.util.NOPMetricsGroup;
import org.apache.fluss.rpc.TestingGatewayService;
@@ -52,6 +53,7 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
import static org.apache.fluss.utils.NetUtils.getAvailablePort;
@@ -255,6 +257,30 @@ final class NettyClientTest {
assertThat(nettyClient.connections()).isEmpty();
}
+ /**
+ * A request whose closed check passed right before the client closed must fail instead of
+ * hanging: close() may have finished before the request created its connection, so the
+ * connection is created on an already terminated event loop.
+ */
+ @Test
+ void testConnectionCreatedAfterCloseFailsRequests() throws Exception {
+ nettyClient.close();
+
+ // simulates a sendRequest that passed the isClosed check right before close() started
+ ServerConnection connection = nettyClient.getOrCreateConnection(serverNode);
+ CompletableFuture<ApiMessage> response =
+ connection.send(
+ ApiKeys.API_VERSIONS,
+ new ApiVersionsRequest()
+ .setClientSoftwareName("test")
+ .setClientSoftwareVersion("1.0"));
+
+ assertThatThrownBy(() -> response.get(1, TimeUnit.MINUTES))
+ .isInstanceOf(ExecutionException.class)
+ .hasCauseInstanceOf(NetworkException.class);
+ assertThat(nettyClient.connections()).isEmpty();
+ }
+
private void buildNettyServer(int serverId) throws Exception {
try (NetUtils.Port availablePort = getAvailablePort()) {
serverNode =
diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/NettyServerTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/NettyServerTest.java
index c16f996d7..6e4421870 100644
--- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/NettyServerTest.java
+++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/NettyServerTest.java
@@ -27,7 +27,9 @@ import org.apache.fluss.metrics.groups.MetricGroup;
import org.apache.fluss.metrics.util.NOPMetricsGroup;
import org.apache.fluss.rpc.RpcServer;
import org.apache.fluss.rpc.TestingGatewayService;
+import org.apache.fluss.rpc.messages.ApiMessage;
import org.apache.fluss.rpc.messages.ApiVersionsRequest;
+import org.apache.fluss.rpc.messages.ApiVersionsResponse;
import org.apache.fluss.rpc.metrics.TestingClientMetricGroup;
import org.apache.fluss.rpc.netty.client.NettyClient;
import org.apache.fluss.rpc.protocol.ApiKeys;
@@ -35,9 +37,16 @@ import org.apache.fluss.utils.NetUtils;
import org.junit.jupiter.api.Test;
+import java.net.ConnectException;
+import java.net.Socket;
+import java.time.Duration;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
import static org.apache.fluss.utils.NetUtils.getAvailablePort;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -111,4 +120,94 @@ public class NettyServerTest {
}
}
}
+
+ /**
+ * Requests that were accepted before the server started to shut down still get their responses:
+ * the event loops, and with them the client connections, are only shut down after the request
+ * processors drained, even though the server stopped accepting new connections.
+ */
+ @Test
+ void testCloseRespondsToAcceptedRequests() throws Exception {
+ Configuration conf = new Configuration();
+ conf.setInt(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS, 1);
+ AtomicBoolean blockRequests = new AtomicBoolean(false);
+ CountDownLatch requestBlocked = new CountDownLatch(1);
+ CountDownLatch releaseRequest = new CountDownLatch(1);
+ TestingGatewayService service =
+ new TestingGatewayService() {
+ @Override
+ public CompletableFuture<ApiVersionsResponse> apiVersions(
+ ApiVersionsRequest request) {
+ if (blockRequests.get()) {
+ requestBlocked.countDown();
+ try {
+ releaseRequest.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return super.apiVersions(request);
+ }
+ };
+ MetricGroup metricGroup = NOPMetricsGroup.newInstance();
+ try (NetUtils.Port availablePort = getAvailablePort()) {
+ ServerNode serverNode =
+ new ServerNode(1, "localhost", availablePort.getPort(), ServerType.COORDINATOR);
+ try (NettyServer nettyServer =
+ new NettyServer(
+ conf,
+ Collections.singleton(
+ new Endpoint(
+ serverNode.host(),
+ serverNode.port(),
+ "INTERNAL")),
+ service,
+ metricGroup,
+ RequestsMetrics.createCoordinatorServerRequestMetrics(
+ metricGroup));
+ NettyClient nettyClient =
+ new NettyClient(
+ new Configuration(), TestingClientMetricGroup.newInstance())) {
+ nettyServer.start();
+ ApiVersionsRequest request =
+ new ApiVersionsRequest()
+ .setClientSoftwareName("test")
+ .setClientSoftwareVersion("1.0");
+ // establish the connection
+ nettyClient.sendRequest(serverNode, ApiKeys.API_VERSIONS, request).get();
+
+ Endpoint bindEndpoint = nettyServer.getBindEndpoints().get(0);
+ new Socket(bindEndpoint.getHost(), bindEndpoint.getPort()).close();
+
+ blockRequests.set(true);
+ try {
+ CompletableFuture<ApiMessage> response =
+ nettyClient.sendRequest(serverNode, ApiKeys.API_VERSIONS, request);
+ requestBlocked.await();
+
+ CompletableFuture<Void> closeFuture = nettyServer.closeAsync();
+ // the server no longer accepts new connections ...
+ retry(
+ Duration.ofMinutes(1),
+ () ->
+ assertThatThrownBy(
+ () ->
+ new Socket(
+ bindEndpoint.getHost(),
+ bindEndpoint.getPort())
+ .close())
+ .isInstanceOf(ConnectException.class));
+ assertThat(closeFuture).isNotDone();
+
+ // ... but still responds to the request it accepted before
+ releaseRequest.countDown();
+ assertThat(response.get()).isInstanceOf(ApiVersionsResponse.class);
+ closeFuture.get();
+ } finally {
+ // never leave the request processor blocked, even if an assertion failed
+ releaseRequest.countDown();
+ }
+ }
+ }
+ }
}
diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/RequestChannelTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/RequestChannelTest.java
index b433d0082..cfa1cb0ee 100644
--- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/RequestChannelTest.java
+++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/protocol/RequestChannelTest.java
@@ -54,6 +54,7 @@ import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -254,6 +255,39 @@ public class RequestChannelTest {
assertThat(testChannel.isAutoRead()).isFalse();
}
+ /**
+ * Test that backpressure transitions do not throw when a channel's event loop is already
+ * shutting down (e.g. during server shutdown) and rejects new tasks.
+ */
+ @Test
+ void testBackpressureIgnoresRejectedEventLoopTasks() {
+ int backpressureThreshold = 10;
+ RequestChannel channel = new RequestChannel(backpressureThreshold);
+ TestChannel testChannel = new TestChannel();
+ channel.registerChannel(testChannel);
+ testChannel.eventLoop.rejectTasks(true);
+
+ // pausing the channel is rejected by its event loop, but putRequest must not throw
+ for (int i = 0; i < backpressureThreshold; i++) {
+ channel.putRequest(createTestRequest(i));
+ }
+ assertThat(testChannel.isAutoRead()).isTrue();
+
+ // resuming the channel is rejected as well, but pollRequest must not throw
+ for (int i = 0; i < backpressureThreshold; i++) {
+ assertThat(channel.pollRequest(10)).isNotNull();
+ }
+ assertThat(channel.requestsCount()).isZero();
+
+ // a real event loop never accepts tasks again after shutting down; this only verifies
+ // that the skipped tasks did not leave the backpressure state machine stuck
+ testChannel.eventLoop.rejectTasks(false);
+ for (int i = 0; i < backpressureThreshold; i++) {
+ channel.putRequest(createTestRequest(i));
+ }
+ assertThat(testChannel.isAutoRead()).isFalse();
+ }
+
@Test
void testZeroBackpressureThresholdDoesNotSpin() throws Exception {
RequestChannel channel = new RequestChannel(0);
@@ -686,6 +720,7 @@ public class RequestChannelTest {
/** Test EventLoop that executes tasks immediately in the current thread for testing. */
private static class TestEventLoop extends SingleThreadEventExecutor implements EventLoop {
private final AtomicBoolean blockNextExecution = new AtomicBoolean();
+ private final AtomicBoolean rejectTasks = new AtomicBoolean();
private final CountDownLatch taskBlocked = new CountDownLatch(1);
private final CountDownLatch blockedTaskRelease = new CountDownLatch(1);
@@ -693,6 +728,11 @@ public class RequestChannelTest {
super(null, new DefaultThreadFactory("test"), false);
}
+ /** Simulates an event loop that is shutting down and rejects new tasks. */
+ void rejectTasks(boolean reject) {
+ rejectTasks.set(reject);
+ }
+
void blockNextTask() {
blockNextExecution.set(true);
}
@@ -712,6 +752,9 @@ public class RequestChannelTest {
@Override
public void execute(Runnable task) {
+ if (rejectTasks.get()) {
+ throw new RejectedExecutionException("event executor terminated");
+ }
if (blockNextExecution.compareAndSet(true, false)) {
taskBlocked.countDown();
try {