You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by le...@apache.org on 2026/09/11 08:05:07 UTC
(fluss) branch main updated: [server] Add per-tablet-server sender retry framework (#3391)
This is an automated email from the ASF dual-hosted git repository.
leekeiabstraction 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 ee584f899 [server] Add per-tablet-server sender retry framework (#3391)
ee584f899 is described below
commit ee584f89929e8b457e7a1fe3d2cb7def7141927b
Author: Yang Guo <ji...@outlook.com>
AuthorDate: Fri Sep 11 16:05:02 2026 +0800
[server] Add per-tablet-server sender retry framework (#3391)
* fix: stop-replica-failed
* fix: stop-replica-failed
* fix: stop-replica-failed
* feat: per tablet server sender thread
* fix conflict
* [server] Isolate per-tablet-server sender framework
Keep the generic per-tablet-server queue, sender, retry, metrics, and lifecycle components while restoring concrete StopReplica handling to the main branch behavior.
Co-Authored-By: Codex <no...@openai.com>
AI-Model: gpt-5
AI-Contributed/Feature: 875/875
AI-Contributed/UT: 1342/1342
* [server] Harden per-tablet-server sender retries
Restrict retries to transport failures, invalidate timed-out RPC connections before retrying, and make gateway access thread-safe. Validate retry durations and keep sender liveness metrics accurate.
Co-Authored-By: Codex <no...@openai.com>
AI-Model: gpt-5
AI-Contributed/Feature: 223/223
AI-Contributed/UT: 305/305
* [server] Address sender retry review comments
Retain the queue head while connection invalidation is pending or retried. Clarify control-request configuration and metric names, and document the new options.
Co-Authored-By: Codex <no...@openai.com>
AI-Model: gpt-5
AI-Contributed/Feature: 103/103
AI-Contributed/UT: 126/126
---------
Co-authored-by: Codex <no...@openai.com>
---
.../org/apache/fluss/config/ConfigOptions.java | 18 +
.../org/apache/fluss/config/FlussConfigUtils.java | 2 +
.../java/org/apache/fluss/metrics/MetricNames.java | 6 +
.../fluss/utils/concurrent/ShutdownableThread.java | 11 +
.../apache/fluss/config/FlussConfigUtilsTest.java | 22 +
.../apache/fluss/metrics/util/TestMetricGroup.java | 21 +-
.../coordinator/CoordinatorChannelManager.java | 170 +++++-
.../server/coordinator/CoordinatorServer.java | 18 +-
.../channel/ControlRequestSendThread.java | 321 +++++++++++
.../server/coordinator/channel/QueueItem.java | 72 +++
.../channel/TabletServerChannelState.java | 63 +++
.../fluss/server/utils/RpcGatewayManager.java | 35 +-
.../coordinator/CoordinatorChannelManagerTest.java | 33 +-
.../coordinator/TestCoordinatorChannelManager.java | 8 +-
.../channel/ControlRequestSendThreadTest.java | 602 +++++++++++++++++++++
.../statemachine/ReplicaStateMachineTest.java | 6 +-
.../statemachine/TableBucketStateMachineTest.java | 5 +-
.../fluss/server/utils/RpcGatewayManagerTest.java | 101 ++++
website/docs/maintenance/configuration.md | 2 +
19 files changed, 1488 insertions(+), 28 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index 173173103..14b019853 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -569,6 +569,24 @@ public class ConfigOptions {
"The interval for cleaning up expired producer offsets "
+ "and orphan files in remote storage. Default is 1 hour.");
+ public static final ConfigOption<Duration> COORDINATOR_CONTROL_REQUEST_RETRY_BACKOFF =
+ key("coordinator.control-request.retry-backoff")
+ .durationType()
+ .defaultValue(Duration.ofMillis(100))
+ .withDescription(
+ "The backoff duration the coordinator waits before retrying a "
+ + "control-plane request to a tablet server after a "
+ + "transient RPC-layer failure.");
+
+ public static final ConfigOption<Duration> COORDINATOR_CONTROL_REQUEST_TIMEOUT =
+ key("coordinator.control-request.timeout")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(30))
+ .withDescription(
+ "The timeout the sender thread waits for a response to a "
+ + "control-plane request before treating it as failed "
+ + "and retrying.");
+
// ------------------------------------------------------------------------
// ConfigOptions for Tablet Server
// ------------------------------------------------------------------------
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
index cf27f2e31..09ed60d63 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
@@ -124,6 +124,8 @@ public class FlussConfigUtils {
public static void validateCoordinatorConfigs(Configuration conf) {
validateServerConfigs(conf);
+ validMinDuration(conf, ConfigOptions.COORDINATOR_CONTROL_REQUEST_RETRY_BACKOFF, 1);
+ validMinDuration(conf, ConfigOptions.COORDINATOR_CONTROL_REQUEST_TIMEOUT, 1);
}
public static void validateTabletConfigs(Configuration conf) {
diff --git a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
index 766e0e9ab..000baef27 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
@@ -47,6 +47,12 @@ public class MetricNames {
public static final String KV_LEADER_REPLICA_CAPACITY = "kvLeaderReplicaCapacity";
public static final String REPLICAS_TO_DELETE_COUNT = "replicasToDeleteCount";
public static final String PENDING_LEADER_ACTIVATION_COUNT = "pendingLeaderActivationCount";
+ // for coordinator sender (per-tablet-server control request sender threads)
+ public static final String SENDER_QUEUE_SIZE = "senderQueueSize";
+ public static final String SENDER_QUEUE_TIME_MS = "senderQueueTimeMs";
+ public static final String SENDER_RETRY_COUNT = "senderRetryCount";
+ public static final String SENDER_STALE_DROP_COUNT = "senderStaleDropCount";
+ public static final String SENDER_ALIVE = "senderAlive";
// for coordinator event processor
public static final String EVENT_QUEUE_SIZE = "eventQueueSize";
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/concurrent/ShutdownableThread.java b/fluss-common/src/main/java/org/apache/fluss/utils/concurrent/ShutdownableThread.java
index 54a202fee..9c2bc308a 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/concurrent/ShutdownableThread.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/concurrent/ShutdownableThread.java
@@ -21,6 +21,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
/** An abstract thread that is shutdownable . */
public abstract class ShutdownableThread extends Thread {
@@ -110,6 +111,16 @@ public abstract class ShutdownableThread extends Thread {
log.info("Stopped");
}
+ /**
+ * Causes the current thread to wait until the shutdown is initiated, or the specified waiting
+ * time elapses.
+ */
+ protected void pause(long timeout, TimeUnit unit) throws InterruptedException {
+ if (shutdownInitiated.await(timeout, unit)) {
+ log.trace("shutdownInitiated latch count reached zero. Shutdown called.");
+ }
+ }
+
public boolean isRunning() {
return !isShutdownInitiated();
}
diff --git a/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java b/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java
index 61a00c01d..1a136167e 100644
--- a/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java
@@ -215,6 +215,28 @@ class FlussConfigUtilsTest {
validateCoordinatorConfigs(conf);
}
+ @Test
+ void testValidateCoordinatorRequestDurations() {
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.REMOTE_DATA_DIR, "s3://bucket/path");
+
+ conf.set(ConfigOptions.COORDINATOR_CONTROL_REQUEST_RETRY_BACKOFF, Duration.ZERO);
+ assertThatThrownBy(() -> validateCoordinatorConfigs(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.COORDINATOR_CONTROL_REQUEST_RETRY_BACKOFF.key())
+ .hasMessageContaining("must be greater than or equal 1 ms");
+
+ conf.set(ConfigOptions.COORDINATOR_CONTROL_REQUEST_RETRY_BACKOFF, Duration.ofMillis(1));
+ conf.set(ConfigOptions.COORDINATOR_CONTROL_REQUEST_TIMEOUT, Duration.ZERO);
+ assertThatThrownBy(() -> validateCoordinatorConfigs(conf))
+ .isInstanceOf(IllegalConfigurationException.class)
+ .hasMessageContaining(ConfigOptions.COORDINATOR_CONTROL_REQUEST_TIMEOUT.key())
+ .hasMessageContaining("must be greater than or equal 1 ms");
+
+ conf.set(ConfigOptions.COORDINATOR_CONTROL_REQUEST_TIMEOUT, Duration.ofMillis(1));
+ validateCoordinatorConfigs(conf);
+ }
+
@Test
void testValidateHistoricalLookupCacheConfigs() {
Configuration conf = new Configuration();
diff --git a/fluss-common/src/test/java/org/apache/fluss/metrics/util/TestMetricGroup.java b/fluss-common/src/test/java/org/apache/fluss/metrics/util/TestMetricGroup.java
index 785574037..7b83d20a4 100644
--- a/fluss-common/src/test/java/org/apache/fluss/metrics/util/TestMetricGroup.java
+++ b/fluss-common/src/test/java/org/apache/fluss/metrics/util/TestMetricGroup.java
@@ -22,10 +22,12 @@ import org.apache.fluss.metrics.Counter;
import org.apache.fluss.metrics.Gauge;
import org.apache.fluss.metrics.Histogram;
import org.apache.fluss.metrics.Meter;
+import org.apache.fluss.metrics.Metric;
import org.apache.fluss.metrics.SimpleCounter;
import org.apache.fluss.metrics.groups.MetricGroup;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.function.BiFunction;
@@ -37,6 +39,7 @@ public class TestMetricGroup implements MetricGroup {
private final Map<String, String> variables;
private final BiFunction<String, Optional<CharacterFilter>, String> metricIdentifierFunction;
private final BiFunction<CharacterFilter, Optional<Character>, String> logicalScopeFunction;
+ private final Map<String, Metric> metrics = new HashMap<>();
public TestMetricGroup(
String[] scopeComponents,
@@ -49,32 +52,48 @@ public class TestMetricGroup implements MetricGroup {
this.logicalScopeFunction = logicalScopeFunction;
}
+ /** Creates a default {@link TestMetricGroup} suitable for unit tests. */
+ public static TestMetricGroup createTestMetricGroup() {
+ return newBuilder().build();
+ }
+
public static TestMetricGroupBuilder newBuilder() {
return new TestMetricGroupBuilder();
}
+ /** Returns a metric previously registered under the given name. */
+ public Metric getMetric(String name) {
+ return metrics.get(name);
+ }
+
@Override
public Counter counter(String name) {
- return new SimpleCounter();
+ SimpleCounter counter = new SimpleCounter();
+ metrics.put(name, counter);
+ return counter;
}
@Override
public <C extends Counter> C counter(String name, C counter) {
+ metrics.put(name, counter);
return counter;
}
@Override
public <T, G extends Gauge<T>> G gauge(String name, G gauge) {
+ metrics.put(name, gauge);
return gauge;
}
@Override
public <H extends Histogram> H histogram(String name, H histogram) {
+ metrics.put(name, histogram);
return histogram;
}
@Override
public <M extends Meter> M meter(String name, M meter) {
+ metrics.put(name, meter);
return meter;
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorChannelManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorChannelManager.java
index e095935ae..fe5bf4273 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorChannelManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorChannelManager.java
@@ -20,6 +20,9 @@ package org.apache.fluss.server.coordinator;
import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.cluster.ServerNode;
import org.apache.fluss.cluster.ServerType;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metrics.MetricNames;
+import org.apache.fluss.metrics.groups.MetricGroup;
import org.apache.fluss.rpc.RpcClient;
import org.apache.fluss.rpc.gateway.TabletServerGateway;
import org.apache.fluss.rpc.messages.ApiMessage;
@@ -35,25 +38,36 @@ import org.apache.fluss.rpc.messages.StopReplicaRequest;
import org.apache.fluss.rpc.messages.StopReplicaResponse;
import org.apache.fluss.rpc.messages.UpdateMetadataRequest;
import org.apache.fluss.rpc.messages.UpdateMetadataResponse;
+import org.apache.fluss.rpc.protocol.ApiKeys;
+import org.apache.fluss.server.coordinator.channel.ControlRequestSendThread;
+import org.apache.fluss.server.coordinator.channel.QueueItem;
+import org.apache.fluss.server.coordinator.channel.TabletServerChannelState;
import org.apache.fluss.server.utils.RpcGatewayManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.concurrent.NotThreadSafe;
+import javax.annotation.Nullable;
import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.BiConsumer;
+import java.util.function.Function;
+import java.util.function.IntSupplier;
import static org.apache.fluss.utils.Preconditions.checkState;
/**
- * Using by coordinator server. It's a manager to manage the rpc channels to tablet servers and send
- * request to the servers.
+ * Used by the coordinator server to manage RPC channels to tablet servers and send requests.
+ * Mutations are guarded by {@code channelLock} so the metric reporter thread can safely read queue
+ * sizes. Mirrors Kafka's {@code ControllerChannelManager} which uses a {@code brokerLock} for the
+ * same purpose.
*/
-@NotThreadSafe
public class CoordinatorChannelManager {
private static final Logger LOG = LoggerFactory.getLogger(CoordinatorChannelManager.class);
@@ -61,30 +75,96 @@ public class CoordinatorChannelManager {
/** A manager for the rpc gateways to tablet servers. */
private final RpcGatewayManager<TabletServerGateway> rpcGatewayManager;
- public CoordinatorChannelManager(RpcClient rpcClient) {
+ private final IntSupplier epochSupplier;
+ private final Configuration conf;
+ private final MetricGroup coordinatorMetricGroup;
+
+ private final Object channelLock = new Object();
+ private final Map<Integer, TabletServerChannelState> channelStates = new HashMap<>();
+
+ public CoordinatorChannelManager(
+ RpcClient rpcClient,
+ IntSupplier epochSupplier,
+ Configuration conf,
+ MetricGroup coordinatorMetricGroup) {
this.rpcGatewayManager = new RpcGatewayManager<>(rpcClient, TabletServerGateway.class);
+ this.epochSupplier = epochSupplier;
+ this.conf = conf;
+ this.coordinatorMetricGroup = coordinatorMetricGroup;
}
public void startup(Collection<ServerNode> serverNodes) {
for (ServerNode serverNode : serverNodes) {
- addTabletServer(serverNode);
+ addNewTabletServer(serverNode);
+ }
+ synchronized (channelLock) {
+ for (TabletServerChannelState state : channelStates.values()) {
+ startSendThread(state);
+ }
}
}
public void close() throws Exception {
+ shutdown();
rpcGatewayManager.close();
}
+ /** Adds a tablet server and immediately starts its sender thread (runtime addition). */
public void addTabletServer(ServerNode serverNode) {
- // add new tablet server to the channel manager
+ addNewTabletServer(serverNode);
+ synchronized (channelLock) {
+ TabletServerChannelState state = channelStates.get(serverNode.id());
+ if (state != null) {
+ startSendThread(state);
+ }
+ }
+ }
+
+ private void addNewTabletServer(ServerNode serverNode) {
checkState(
serverNode.serverType().equals(ServerType.TABLET_SERVER),
"The server type should be TABLET_SERVER, but was " + serverNode.serverType());
rpcGatewayManager.addServer(serverNode);
+
+ int id = serverNode.id();
+ synchronized (channelLock) {
+ if (channelStates.containsKey(id)) {
+ return;
+ }
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+
+ MetricGroup tsGroup =
+ coordinatorMetricGroup.addGroup("tablet_server_id", String.valueOf(id));
+ tsGroup.gauge(MetricNames.SENDER_QUEUE_SIZE, queue::size);
+
+ ControlRequestSendThread thread =
+ new ControlRequestSendThread(
+ id,
+ queue,
+ () -> rpcGatewayManager.getRpcGateway(id),
+ () -> rpcGatewayManager.disconnectServer(id),
+ epochSupplier,
+ conf,
+ tsGroup);
+ channelStates.put(id, new TabletServerChannelState(queue, thread, tsGroup));
+ }
+ }
+
+ private void startSendThread(TabletServerChannelState state) {
+ ControlRequestSendThread thread = state.getSendThread();
+ if (thread.getState() == Thread.State.NEW) {
+ thread.start();
+ }
}
public void removeTabletServer(Integer serverId) {
+ TabletServerChannelState state;
+ synchronized (channelLock) {
+ state = channelStates.remove(serverId);
+ }
+ teardownChannelState(serverId, state);
+
rpcGatewayManager
.removeServer(serverId)
.exceptionally(
@@ -97,6 +177,39 @@ public class CoordinatorChannelManager {
});
}
+ /** Shuts down all per-tablet-server sender threads and deregisters their metrics. */
+ public void shutdown() {
+ Map<Integer, TabletServerChannelState> statesToTeardown;
+ synchronized (channelLock) {
+ statesToTeardown = new HashMap<>(channelStates);
+ channelStates.clear();
+ }
+ for (Map.Entry<Integer, TabletServerChannelState> entry : statesToTeardown.entrySet()) {
+ teardownChannelState(entry.getKey(), entry.getValue());
+ }
+ }
+
+ private void teardownChannelState(int serverId, @Nullable TabletServerChannelState state) {
+ if (state == null) {
+ return;
+ }
+ try {
+ try {
+ state.getSendThread().shutdown();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn(
+ "Interrupted while shutting down sender thread for tabletServer {}",
+ serverId,
+ e);
+ }
+ state.getQueue().clear();
+ state.getMetricGroup().close();
+ } catch (Throwable t) {
+ LOG.error("Error tearing down channel state for tabletServer {}", serverId, t);
+ }
+ }
+
/** Send NotifyLeaderAndIsr request to the server and handle the response. */
public void sendBucketLeaderAndIsrRequest(
int receiveServerId,
@@ -121,6 +234,49 @@ public class CoordinatorChannelManager {
responseConsumer);
}
+ /**
+ * Enqueues a control-plane request for the target tablet server. The per-tablet-server sender
+ * retries transport failures and timeouts until the request succeeds or the channel is removed.
+ * Other failures are completed through the response consumer without retrying.
+ *
+ * @return whether the request was enqueued
+ */
+ protected <ResponseT extends ApiMessage> boolean enqueueRequest(
+ int targetServerId,
+ ApiKeys apiKey,
+ int coordinatorEpoch,
+ Function<TabletServerGateway, CompletableFuture<ResponseT>> requestSender,
+ @Nullable BiConsumer<ResponseT, ? super Throwable> responseConsumer) {
+ TabletServerChannelState state;
+ synchronized (channelLock) {
+ state = channelStates.get(targetServerId);
+ }
+ if (state == null) {
+ LOG.warn(
+ "Cannot enqueue {} for tablet server {} because its channel does not exist.",
+ apiKey,
+ targetServerId);
+ return false;
+ }
+
+ try {
+ state.getQueue()
+ .put(
+ new QueueItem<>(
+ apiKey,
+ requestSender,
+ responseConsumer,
+ coordinatorEpoch,
+ System.currentTimeMillis()));
+ return true;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn(
+ "Interrupted while enqueueing {} for tablet server {}", apiKey, targetServerId);
+ return false;
+ }
+ }
+
/** Send UpdateMetadataRequest to the server and handle the response. */
public void sendUpdateMetadataRequest(
int receiveServerId,
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
index a2f9b4fad..691530f54 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
@@ -328,7 +328,18 @@ public class CoordinatorServer extends ServerBase {
this.clientMetricGroup = new ClientMetricGroup(metricRegistry, SERVER_NAME);
this.rpcClient = RpcClient.create(conf, clientMetricGroup);
- this.coordinatorChannelManager = new CoordinatorChannelManager(rpcClient);
+ // start coordinator event processor after we register coordinator leader to zk
+ // so that the event processor can get the coordinator leader node from zk during
+ // start up. In HA for coordinator server, the processor also need to know the leader
+ // node during start up
+ CoordinatorContext coordinatorContext = new CoordinatorContext(zkEpoch);
+
+ this.coordinatorChannelManager =
+ new CoordinatorChannelManager(
+ rpcClient,
+ coordinatorContext::getCoordinatorEpoch,
+ conf,
+ serverMetricGroup);
this.autoPartitionManager =
new AutoPartitionManager(
@@ -339,11 +350,6 @@ public class CoordinatorServer extends ServerBase {
replicaCapacityController);
autoPartitionManager.start();
- // start coordinator event processor after we register coordinator leader to zk
- // so that the event processor can get the coordinator leader node from zk during
- // start up. In HA for coordinator server, the processor also need to know the leader
- // node during start up
- CoordinatorContext coordinatorContext = new CoordinatorContext(zkEpoch);
this.coordinatorEventProcessor =
new CoordinatorEventProcessor(
zkClient,
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/ControlRequestSendThread.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/ControlRequestSendThread.java
new file mode 100644
index 000000000..40293a3df
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/ControlRequestSendThread.java
@@ -0,0 +1,321 @@
+/*
+ * 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.fluss.server.coordinator.channel;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.DisconnectException;
+import org.apache.fluss.exception.NetworkException;
+import org.apache.fluss.metrics.Counter;
+import org.apache.fluss.metrics.DescriptiveStatisticsHistogram;
+import org.apache.fluss.metrics.Histogram;
+import org.apache.fluss.metrics.MetricNames;
+import org.apache.fluss.metrics.groups.MetricGroup;
+import org.apache.fluss.rpc.gateway.TabletServerGateway;
+import org.apache.fluss.rpc.messages.ApiMessage;
+import org.apache.fluss.utils.concurrent.ShutdownableThread;
+
+import java.util.Optional;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.IntSupplier;
+import java.util.function.Supplier;
+
+import static org.apache.fluss.utils.ExceptionUtils.stripCompletionException;
+import static org.apache.fluss.utils.ExceptionUtils.stripExecutionException;
+
+/**
+ * Per-tablet-server sender thread that drains the control-plane request queue and retries on
+ * transient RPC failures.
+ *
+ * <p>Each invocation of {@link #doWork()} takes one {@link QueueItem} from the queue. Stale items
+ * (whose {@code coordinatorEpoch} is less than the current epoch) are dropped immediately.
+ * Otherwise the item is sent to the tablet server via the gateway, retrying with a configurable
+ * backoff after transport failures until the send succeeds or the thread is shut down. Before a
+ * retry, the old connection is closed so its in-flight request cannot accumulate in the RPC layer.
+ * Explicit server rejections and local request construction failures are completed through the
+ * callback without retrying, allowing the next FIFO item to make progress.
+ *
+ * <p>The callback is invoked outside the retry loop so a response-handler failure does not retry a
+ * request that the tablet server has already completed.
+ */
+public class ControlRequestSendThread extends ShutdownableThread {
+
+ private static final int HISTOGRAM_WINDOW_SIZE = 100;
+
+ private final int tabletServerId;
+ private final BlockingQueue<QueueItem<?>> queue;
+ private final Supplier<Optional<TabletServerGateway>> gatewaySupplier;
+ private final Supplier<CompletableFuture<Void>> connectionInvalidator;
+ private final IntSupplier epochSupplier;
+ private final long backoffMs;
+ private final long requestTimeoutMs;
+
+ private final Histogram queueTimeMsHistogram;
+ private final Counter retryCount;
+ private final Counter staleDropCount;
+ private final AtomicInteger aliveFlag;
+
+ private final AtomicReference<CompletableFuture<? extends ApiMessage>> inFlight =
+ new AtomicReference<>();
+
+ public ControlRequestSendThread(
+ int tabletServerId,
+ BlockingQueue<QueueItem<?>> queue,
+ Supplier<Optional<TabletServerGateway>> gatewaySupplier,
+ Supplier<CompletableFuture<Void>> connectionInvalidator,
+ IntSupplier epochSupplier,
+ Configuration conf,
+ MetricGroup metricGroup) {
+ super("coordinator-control-request-sender-" + tabletServerId, true);
+ this.tabletServerId = tabletServerId;
+ this.queue = queue;
+ this.gatewaySupplier = gatewaySupplier;
+ this.connectionInvalidator = connectionInvalidator;
+ this.epochSupplier = epochSupplier;
+ this.backoffMs =
+ conf.get(ConfigOptions.COORDINATOR_CONTROL_REQUEST_RETRY_BACKOFF).toMillis();
+ this.requestTimeoutMs =
+ conf.get(ConfigOptions.COORDINATOR_CONTROL_REQUEST_TIMEOUT).toMillis();
+
+ this.queueTimeMsHistogram =
+ metricGroup.histogram(
+ MetricNames.SENDER_QUEUE_TIME_MS,
+ new DescriptiveStatisticsHistogram(HISTOGRAM_WINDOW_SIZE));
+ this.retryCount = metricGroup.counter(MetricNames.SENDER_RETRY_COUNT);
+ this.staleDropCount = metricGroup.counter(MetricNames.SENDER_STALE_DROP_COUNT);
+ this.aliveFlag = new AtomicInteger(0);
+ metricGroup.gauge(MetricNames.SENDER_ALIVE, aliveFlag::get);
+ }
+
+ @Override
+ public void run() {
+ aliveFlag.set(1);
+ try {
+ super.run();
+ } finally {
+ aliveFlag.set(0);
+ }
+ }
+
+ @Override
+ public boolean initiateShutdown() {
+ boolean initiated = super.initiateShutdown();
+ if (initiated) {
+ CompletableFuture<? extends ApiMessage> f = inFlight.getAndSet(null);
+ if (f != null) {
+ f.cancel(true);
+ }
+ }
+ return initiated;
+ }
+
+ @Override
+ public void doWork() throws Exception {
+ try {
+ QueueItem<?> item = queue.take();
+ queueTimeMsHistogram.update(System.currentTimeMillis() - item.getEnqueueTimeMs());
+
+ int currentEpoch = epochSupplier.getAsInt();
+ if (item.getCoordinatorEpoch() < currentEpoch) {
+ staleDropCount.inc();
+ log.warn(
+ "Dropping stale {} for tabletServer {}: itemEpoch={} < currentEpoch={}",
+ item.getApiKey(),
+ tabletServerId,
+ item.getCoordinatorEpoch(),
+ currentEpoch);
+ return;
+ }
+
+ send(item);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private <ResponseT extends ApiMessage> void send(QueueItem<ResponseT> item)
+ throws InterruptedException {
+ while (isRunning()) {
+ Optional<TabletServerGateway> gatewayOpt;
+ try {
+ gatewayOpt = gatewaySupplier.get();
+ } catch (RuntimeException e) {
+ Throwable failure = unwrapFailure(e);
+ log.error(
+ "Failed to resolve the gateway for {} to tabletServer {}; "
+ + "the request will not be retried",
+ item.getApiKey(),
+ tabletServerId,
+ failure);
+ invokeCallback(item, null, failure);
+ return;
+ }
+ if (!gatewayOpt.isPresent()) {
+ retryCount.inc();
+ backoff();
+ continue;
+ }
+
+ CompletableFuture<ResponseT> future = null;
+ ResponseT response;
+ Throwable failure;
+ try {
+ future = item.send(gatewayOpt.get());
+ inFlight.set(future);
+ response = future.get(requestTimeoutMs, TimeUnit.MILLISECONDS);
+ invokeCallback(item, response, null);
+ return;
+ } catch (TimeoutException e) {
+ failure = e;
+ } catch (ExecutionException e) {
+ failure = unwrapFailure(e);
+ } catch (RuntimeException e) {
+ failure = unwrapFailure(e);
+ } finally {
+ if (future != null) {
+ inFlight.compareAndSet(future, null);
+ }
+ }
+
+ if (!isRunning()) {
+ return;
+ }
+
+ if (!isRetriableTransportFailure(failure)) {
+ log.error(
+ "Failed to send {} to tabletServer {}; the failure is not retriable",
+ item.getApiKey(),
+ tabletServerId,
+ failure);
+ invokeCallback(item, null, failure);
+ return;
+ }
+
+ if (!invalidateConnection(item)) {
+ return;
+ }
+
+ retryCount.inc();
+ log.warn(
+ "Failed to send {} to tabletServer {}; will retry after {}ms",
+ item.getApiKey(),
+ tabletServerId,
+ backoffMs,
+ failure);
+ backoff();
+ }
+ }
+
+ private boolean invalidateConnection(QueueItem<?> item) throws InterruptedException {
+ CompletableFuture<Void> invalidationFuture = null;
+ while (isRunning()) {
+ if (invalidationFuture == null) {
+ try {
+ invalidationFuture = connectionInvalidator.get();
+ } catch (RuntimeException e) {
+ logInvalidationFailure(item, unwrapFailure(e));
+ retryCount.inc();
+ backoff();
+ continue;
+ }
+ }
+
+ try {
+ invalidationFuture.get(requestTimeoutMs, TimeUnit.MILLISECONDS);
+ return true;
+ } catch (TimeoutException e) {
+ log.warn(
+ "Timed out waiting to invalidate the connection to tabletServer {} after "
+ + "{}; the queue head will be retained and invalidation will be "
+ + "checked again after {}ms",
+ tabletServerId,
+ item.getApiKey(),
+ backoffMs,
+ e);
+ } catch (ExecutionException e) {
+ logInvalidationFailure(item, unwrapFailure(e));
+ invalidationFuture = null;
+ } catch (RuntimeException e) {
+ logInvalidationFailure(item, unwrapFailure(e));
+ invalidationFuture = null;
+ }
+
+ retryCount.inc();
+ backoff();
+ }
+ return false;
+ }
+
+ private void logInvalidationFailure(QueueItem<?> item, Throwable failure) {
+ log.warn(
+ "Failed to invalidate the connection to tabletServer {} after {}; the queue head "
+ + "will be retained and invalidation will be retried after {}ms",
+ tabletServerId,
+ item.getApiKey(),
+ backoffMs,
+ failure);
+ }
+
+ private static Throwable unwrapFailure(Throwable failure) {
+ Throwable unwrapped = failure;
+ Throwable previous;
+ do {
+ previous = unwrapped;
+ unwrapped = stripCompletionException(stripExecutionException(unwrapped));
+ } while (unwrapped != previous);
+
+ if (unwrapped instanceof Error) {
+ throw (Error) unwrapped;
+ }
+ return unwrapped;
+ }
+
+ private static boolean isRetriableTransportFailure(Throwable failure) {
+ return failure instanceof TimeoutException
+ || failure instanceof org.apache.fluss.exception.TimeoutException
+ || failure instanceof NetworkException
+ || failure instanceof DisconnectException;
+ }
+
+ private <ResponseT extends ApiMessage> void invokeCallback(
+ QueueItem<ResponseT> item, ResponseT response, Throwable failure) {
+ if (item.getCallback() == null) {
+ return;
+ }
+ try {
+ item.getCallback().accept(response, failure);
+ } catch (RuntimeException e) {
+ Throwable callbackFailure = unwrapFailure(e);
+ log.error(
+ "Callback for {} to tabletServer {} threw; the request will not be retried.",
+ item.getApiKey(),
+ tabletServerId,
+ callbackFailure);
+ }
+ }
+
+ private void backoff() throws InterruptedException {
+ pause(backoffMs, TimeUnit.MILLISECONDS);
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/QueueItem.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/QueueItem.java
new file mode 100644
index 000000000..07c995678
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/QueueItem.java
@@ -0,0 +1,72 @@
+/*
+ * 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.fluss.server.coordinator.channel;
+
+import org.apache.fluss.rpc.gateway.TabletServerGateway;
+import org.apache.fluss.rpc.messages.ApiMessage;
+import org.apache.fluss.rpc.protocol.ApiKeys;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+/** An immutable item placed on the per-tablet-server sender queue. */
+public final class QueueItem<ResponseT extends ApiMessage> {
+
+ private final ApiKeys apiKey;
+ private final Function<TabletServerGateway, CompletableFuture<ResponseT>> requestSender;
+ @Nullable private final BiConsumer<ResponseT, ? super Throwable> callback;
+ private final int coordinatorEpoch;
+ private final long enqueueTimeMs;
+
+ public QueueItem(
+ ApiKeys apiKey,
+ Function<TabletServerGateway, CompletableFuture<ResponseT>> requestSender,
+ @Nullable BiConsumer<ResponseT, ? super Throwable> callback,
+ int coordinatorEpoch,
+ long enqueueTimeMs) {
+ this.apiKey = apiKey;
+ this.requestSender = requestSender;
+ this.callback = callback;
+ this.coordinatorEpoch = coordinatorEpoch;
+ this.enqueueTimeMs = enqueueTimeMs;
+ }
+
+ public ApiKeys getApiKey() {
+ return apiKey;
+ }
+
+ public CompletableFuture<ResponseT> send(TabletServerGateway gateway) {
+ return requestSender.apply(gateway);
+ }
+
+ @Nullable
+ public BiConsumer<ResponseT, ? super Throwable> getCallback() {
+ return callback;
+ }
+
+ public int getCoordinatorEpoch() {
+ return coordinatorEpoch;
+ }
+
+ public long getEnqueueTimeMs() {
+ return enqueueTimeMs;
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/TabletServerChannelState.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/TabletServerChannelState.java
new file mode 100644
index 000000000..113129dd0
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/TabletServerChannelState.java
@@ -0,0 +1,63 @@
+/*
+ * 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.fluss.server.coordinator.channel;
+
+import org.apache.fluss.metrics.groups.MetricGroup;
+
+import javax.annotation.concurrent.ThreadSafe;
+
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * Per-tablet-server channel state held by {@link
+ * org.apache.fluss.server.coordinator.CoordinatorChannelManager}: the queue, the sender thread, and
+ * the metric handles needed to deregister the per-TS metrics on remove.
+ */
+@ThreadSafe
+public final class TabletServerChannelState {
+
+ private final BlockingQueue<QueueItem<?>> queue;
+ private final ControlRequestSendThread sendThread;
+
+ /**
+ * The per-TS child {@link MetricGroup}. Held so {@code removeTabletServer} can {@code close()}
+ * it and deregister all per-TS metrics in one shot.
+ */
+ private final MetricGroup metricGroup;
+
+ public TabletServerChannelState(
+ BlockingQueue<QueueItem<?>> queue,
+ ControlRequestSendThread sendThread,
+ MetricGroup metricGroup) {
+ this.queue = queue;
+ this.sendThread = sendThread;
+ this.metricGroup = metricGroup;
+ }
+
+ public BlockingQueue<QueueItem<?>> getQueue() {
+ return queue;
+ }
+
+ public ControlRequestSendThread getSendThread() {
+ return sendThread;
+ }
+
+ public MetricGroup getMetricGroup() {
+ return metricGroup;
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/RpcGatewayManager.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/RpcGatewayManager.java
index c89fca6f7..4a1ae815c 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/RpcGatewayManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/RpcGatewayManager.java
@@ -26,7 +26,7 @@ import org.apache.fluss.utils.concurrent.FutureUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.concurrent.NotThreadSafe;
+import javax.annotation.concurrent.ThreadSafe;
import java.util.HashMap;
import java.util.Map;
@@ -34,7 +34,7 @@ import java.util.Optional;
import java.util.concurrent.CompletableFuture;
/** A manager to manage the rpc gateways to the servers. */
-@NotThreadSafe
+@ThreadSafe
public class RpcGatewayManager<T extends RpcGateway> implements AutoCloseable {
private static final Logger LOG = LoggerFactory.getLogger(RpcGatewayManager.class);
@@ -54,14 +54,14 @@ public class RpcGatewayManager<T extends RpcGateway> implements AutoCloseable {
/**
* Get the rpc gateway of the server with the given id.
*
- * @param serverUid the id of the server
+ * @param serverId the id of the server
* @return the rpc gateway of the server, empty if the server doesn't exist
*/
- public Optional<T> getRpcGateway(int serverUid) {
- if (!serverRpcGateways.containsKey(serverUid)) {
- return Optional.empty();
- }
- return Optional.of(serverRpcGateways.get(serverUid).rpcGateway);
+ public synchronized Optional<T> getRpcGateway(int serverId) {
+ ServerRpcGateway serverRpcGateway = serverRpcGateways.get(serverId);
+ return serverRpcGateway == null
+ ? Optional.empty()
+ : Optional.of(serverRpcGateway.rpcGateway);
}
/**
@@ -69,7 +69,7 @@ public class RpcGatewayManager<T extends RpcGateway> implements AutoCloseable {
* manager. If the server has already existed, it'll remove the already existing server before
* adding the new one.
*/
- public void addServer(ServerNode serverNode) {
+ public synchronized void addServer(ServerNode serverNode) {
int serverId = serverNode.id();
if (serverRpcGateways.containsKey(serverId)) {
// close the already existing server
@@ -95,7 +95,7 @@ public class RpcGatewayManager<T extends RpcGateway> implements AutoCloseable {
* @param serverId the id of the server to be removed
* @return a future to be completed when the disconnection is complete
*/
- public CompletableFuture<Void> removeServer(int serverId) {
+ public synchronized CompletableFuture<Void> removeServer(int serverId) {
ServerRpcGateway serverRpcGateway = serverRpcGateways.remove(serverId);
if (serverRpcGateway != null) {
return rpcClient.disconnect(serverRpcGateway.serverUid);
@@ -103,6 +103,21 @@ public class RpcGatewayManager<T extends RpcGateway> implements AutoCloseable {
return FutureUtils.completedVoidFuture();
}
+ /**
+ * Disconnects the current RPC connection to a server without removing its gateway. A subsequent
+ * request through the gateway will establish a new connection.
+ *
+ * @param serverId the id of the server to disconnect
+ * @return a future completed after the old connection and its in-flight requests are closed
+ */
+ public synchronized CompletableFuture<Void> disconnectServer(int serverId) {
+ ServerRpcGateway serverRpcGateway = serverRpcGateways.get(serverId);
+ if (serverRpcGateway != null) {
+ return rpcClient.disconnect(serverRpcGateway.serverUid);
+ }
+ return FutureUtils.completedVoidFuture();
+ }
+
@Override
public void close() throws Exception {
// do nothing.
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorChannelManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorChannelManagerTest.java
index 31c07d75c..e040fa476 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorChannelManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorChannelManagerTest.java
@@ -20,8 +20,11 @@ package org.apache.fluss.server.coordinator;
import org.apache.fluss.cluster.ServerNode;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.rpc.RpcClient;
+import org.apache.fluss.rpc.messages.ApiVersionsResponse;
import org.apache.fluss.rpc.messages.UpdateMetadataRequest;
import org.apache.fluss.rpc.metrics.TestingClientMetricGroup;
+import org.apache.fluss.rpc.protocol.ApiKeys;
+import org.apache.fluss.server.metrics.group.TestingMetricGroups;
import org.apache.fluss.server.testutils.FlussClusterExtension;
import org.apache.fluss.server.zk.ZooKeeperExtension;
import org.apache.fluss.testutils.common.AllCallbackWrapper;
@@ -32,6 +35,7 @@ import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeUpdateMetadataRequest;
@@ -54,7 +58,10 @@ class CoordinatorChannelManagerTest {
Configuration configuration = new Configuration();
CoordinatorChannelManager coordinatorChannelManager =
new CoordinatorChannelManager(
- RpcClient.create(configuration, TestingClientMetricGroup.newInstance()));
+ RpcClient.create(configuration, TestingClientMetricGroup.newInstance()),
+ () -> 0,
+ configuration,
+ TestingMetricGroups.COORDINATOR_METRICS);
List<ServerNode> tabletServersNode = FLUSS_CLUSTER_EXTENSION.getTabletServerNodes();
// test start up using server 0
@@ -62,11 +69,13 @@ class CoordinatorChannelManagerTest {
coordinatorChannelManager.startup(Collections.singletonList(server0));
// try to send message, should send
checkSendRequest(coordinatorChannelManager, server0.id(), true);
+ checkEnqueueRequest(coordinatorChannelManager, server0.id(), true);
// test remove tablet server
coordinatorChannelManager.removeTabletServer(server0.id());
// now, shouldn't send as we already remove the tablet server
checkSendRequest(coordinatorChannelManager, server0.id(), false);
+ checkEnqueueRequest(coordinatorChannelManager, server0.id(), false);
// test add tablet server
// before add, shouldn't send
@@ -78,10 +87,32 @@ class CoordinatorChannelManagerTest {
// after add the tablet server, should send
// try to send message
checkSendRequest(coordinatorChannelManager, server1.id(), true);
+ checkEnqueueRequest(coordinatorChannelManager, server1.id(), true);
coordinatorChannelManager.close();
}
+ private void checkEnqueueRequest(
+ CoordinatorChannelManager coordinatorChannelManager,
+ int targetServerId,
+ boolean expectCanEnqueue) {
+ AtomicInteger sendFlag = new AtomicInteger(0);
+ boolean enqueued =
+ coordinatorChannelManager.enqueueRequest(
+ targetServerId,
+ ApiKeys.API_VERSIONS,
+ 0,
+ ignored -> {
+ sendFlag.set(1);
+ return CompletableFuture.completedFuture(new ApiVersionsResponse());
+ },
+ (response, throwable) -> sendFlag.set(2));
+
+ assertThat(enqueued).isEqualTo(expectCanEnqueue);
+ int expectedFlag = expectCanEnqueue ? 2 : 0;
+ retry(Duration.ofMinutes(1), () -> assertThat(sendFlag.get()).isEqualTo(expectedFlag));
+ }
+
private void checkSendRequest(
CoordinatorChannelManager coordinatorChannelManager,
int targetServerId,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorChannelManager.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorChannelManager.java
index a25c4f4ce..0a2ccda24 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorChannelManager.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorChannelManager.java
@@ -21,6 +21,7 @@ import org.apache.fluss.config.Configuration;
import org.apache.fluss.rpc.RpcClient;
import org.apache.fluss.rpc.gateway.TabletServerGateway;
import org.apache.fluss.rpc.metrics.TestingClientMetricGroup;
+import org.apache.fluss.server.metrics.group.TestingMetricGroups;
import java.util.Collections;
import java.util.Map;
@@ -36,7 +37,11 @@ public class TestCoordinatorChannelManager extends CoordinatorChannelManager {
}
public TestCoordinatorChannelManager(Map<Integer, TabletServerGateway> gateways) {
- super(RpcClient.create(new Configuration(), TestingClientMetricGroup.newInstance()));
+ super(
+ RpcClient.create(new Configuration(), TestingClientMetricGroup.newInstance()),
+ () -> 0,
+ new Configuration(),
+ TestingMetricGroups.COORDINATOR_METRICS);
this.gateways = gateways;
}
@@ -44,6 +49,7 @@ public class TestCoordinatorChannelManager extends CoordinatorChannelManager {
this.gateways = gateways;
}
+ @Override
protected Optional<TabletServerGateway> getTabletServerGateway(int serverId) {
return Optional.ofNullable(gateways.get(serverId));
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/channel/ControlRequestSendThreadTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/channel/ControlRequestSendThreadTest.java
new file mode 100644
index 000000000..aee06b8c5
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/channel/ControlRequestSendThreadTest.java
@@ -0,0 +1,602 @@
+/*
+ * 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.fluss.server.coordinator.channel;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.NetworkException;
+import org.apache.fluss.metrics.Counter;
+import org.apache.fluss.metrics.Gauge;
+import org.apache.fluss.metrics.MetricNames;
+import org.apache.fluss.metrics.groups.MetricGroup;
+import org.apache.fluss.metrics.util.TestMetricGroup;
+import org.apache.fluss.rpc.gateway.TabletServerGateway;
+import org.apache.fluss.rpc.messages.ApiVersionsResponse;
+import org.apache.fluss.rpc.protocol.ApiKeys;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Proxy;
+import java.time.Duration;
+import java.util.Optional;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.IntSupplier;
+import java.util.function.Supplier;
+
+import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link ControlRequestSendThread}. */
+class ControlRequestSendThreadTest {
+
+ private static final int TABLET_SERVER_ID = 1;
+ private static final int EPOCH = 1;
+
+ private ControlRequestSendThread thread;
+
+ @AfterEach
+ void afterEach() throws InterruptedException {
+ if (thread != null) {
+ thread.initiateShutdown();
+ thread.awaitShutdown();
+ }
+ }
+
+ @Test
+ void testHappyPath() throws Exception {
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+ MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup();
+
+ ApiVersionsResponse response = new ApiVersionsResponse();
+ AtomicInteger invocationCount = new AtomicInteger(0);
+ TabletServerGateway gateway = unusedGateway();
+
+ AtomicReference<Object> callbackResponse = new AtomicReference<>();
+ CountDownLatch callbackLatch = new CountDownLatch(1);
+
+ thread = createThread(queue, () -> Optional.of(gateway), () -> EPOCH, metricGroup);
+ thread.start();
+
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> {
+ invocationCount.incrementAndGet();
+ return CompletableFuture.completedFuture(response);
+ },
+ (resp, err) -> {
+ callbackResponse.set(resp);
+ callbackLatch.countDown();
+ },
+ EPOCH,
+ System.currentTimeMillis()));
+
+ assertThat(callbackLatch.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(callbackResponse.get()).isSameAs(response);
+ assertThat(getCounter(metricGroup, MetricNames.SENDER_RETRY_COUNT).getCount()).isEqualTo(0);
+ assertThat(getCounter(metricGroup, MetricNames.SENDER_STALE_DROP_COUNT).getCount())
+ .isEqualTo(0);
+ assertThat(getGaugeValue(metricGroup, MetricNames.SENDER_ALIVE)).isEqualTo(1);
+ assertThat(invocationCount.get()).isEqualTo(1);
+ }
+
+ @Test
+ void testRetryThenSucceed() throws Exception {
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+ MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup();
+
+ ApiVersionsResponse response = new ApiVersionsResponse();
+ AtomicInteger callCount = new AtomicInteger(0);
+ TabletServerGateway gateway = unusedGateway();
+
+ CountDownLatch callbackLatch = new CountDownLatch(1);
+ AtomicInteger invalidationCount = new AtomicInteger(0);
+
+ thread =
+ createThread(
+ queue,
+ () -> Optional.of(gateway),
+ () -> {
+ invalidationCount.incrementAndGet();
+ return CompletableFuture.completedFuture(null);
+ },
+ () -> EPOCH,
+ metricGroup,
+ Duration.ofSeconds(30));
+ thread.start();
+
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> {
+ if (callCount.incrementAndGet() == 1) {
+ CompletableFuture<ApiVersionsResponse> fail =
+ new CompletableFuture<>();
+ fail.completeExceptionally(new NetworkException("transient error"));
+ return fail;
+ }
+ return CompletableFuture.completedFuture(response);
+ },
+ (resp, err) -> callbackLatch.countDown(),
+ EPOCH,
+ System.currentTimeMillis()));
+
+ assertThat(callbackLatch.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(getCounter(metricGroup, MetricNames.SENDER_RETRY_COUNT).getCount())
+ .isGreaterThanOrEqualTo(1);
+ assertThat(invalidationCount.get()).isEqualTo(1);
+ }
+
+ @Test
+ void testNonRetriableFailureAdvancesQueue() throws Exception {
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+ MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup();
+ TabletServerGateway gateway = unusedGateway();
+
+ IllegalStateException permanentFailure = new IllegalStateException("invalid request");
+ AtomicReference<Throwable> callbackFailure = new AtomicReference<>();
+ AtomicInteger firstRequestInvocations = new AtomicInteger(0);
+ CountDownLatch secondCallbackLatch = new CountDownLatch(1);
+
+ thread = createThread(queue, () -> Optional.of(gateway), () -> EPOCH, metricGroup);
+ thread.start();
+
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> {
+ firstRequestInvocations.incrementAndGet();
+ throw permanentFailure;
+ },
+ (response, failure) -> callbackFailure.set(failure),
+ EPOCH,
+ System.currentTimeMillis()));
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> CompletableFuture.completedFuture(new ApiVersionsResponse()),
+ (response, failure) -> secondCallbackLatch.countDown(),
+ EPOCH,
+ System.currentTimeMillis()));
+
+ assertThat(secondCallbackLatch.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(callbackFailure.get()).isSameAs(permanentFailure);
+ assertThat(firstRequestInvocations.get()).isEqualTo(1);
+ assertThat(getCounter(metricGroup, MetricNames.SENDER_RETRY_COUNT).getCount()).isEqualTo(0);
+ }
+
+ @Test
+ void testTimeoutInvalidatesConnectionBeforeRetry() throws Exception {
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+ MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup();
+ TabletServerGateway gateway = unusedGateway();
+
+ CompletableFuture<ApiVersionsResponse> firstAttempt = new CompletableFuture<>();
+ ApiVersionsResponse response = new ApiVersionsResponse();
+ AtomicInteger invocationCount = new AtomicInteger(0);
+ AtomicInteger invalidationCount = new AtomicInteger(0);
+ CountDownLatch callbackLatch = new CountDownLatch(1);
+
+ thread =
+ createThread(
+ queue,
+ () -> Optional.of(gateway),
+ () -> {
+ invalidationCount.incrementAndGet();
+ firstAttempt.completeExceptionally(
+ new NetworkException("connection invalidated"));
+ return CompletableFuture.completedFuture(null);
+ },
+ () -> EPOCH,
+ metricGroup,
+ Duration.ofMillis(20));
+ thread.start();
+
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored ->
+ invocationCount.incrementAndGet() == 1
+ ? firstAttempt
+ : CompletableFuture.completedFuture(response),
+ (callbackResponse, failure) -> callbackLatch.countDown(),
+ EPOCH,
+ System.currentTimeMillis()));
+
+ assertThat(callbackLatch.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(firstAttempt.isCompletedExceptionally()).isTrue();
+ assertThat(invalidationCount.get()).isEqualTo(1);
+ assertThat(invocationCount.get()).isEqualTo(2);
+ }
+
+ @Test
+ void testConnectionInvalidationFailureRetainsQueueHead() throws Exception {
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+ MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup();
+ TabletServerGateway gateway = unusedGateway();
+ ApiVersionsResponse response = new ApiVersionsResponse();
+ AtomicInteger sendCount = new AtomicInteger();
+ AtomicInteger invalidationCount = new AtomicInteger();
+ AtomicReference<ApiVersionsResponse> callbackResponse = new AtomicReference<>();
+ CountDownLatch callbackLatch = new CountDownLatch(1);
+
+ thread =
+ createThread(
+ queue,
+ () -> Optional.of(gateway),
+ () -> {
+ if (invalidationCount.incrementAndGet() == 1) {
+ CompletableFuture<Void> failure = new CompletableFuture<>();
+ failure.completeExceptionally(
+ new NetworkException("invalidation failed"));
+ return failure;
+ }
+ return CompletableFuture.completedFuture(null);
+ },
+ () -> EPOCH,
+ metricGroup,
+ Duration.ofSeconds(30));
+ thread.start();
+
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> {
+ if (sendCount.incrementAndGet() == 1) {
+ CompletableFuture<ApiVersionsResponse> failure =
+ new CompletableFuture<>();
+ failure.completeExceptionally(
+ new NetworkException("request failed"));
+ return failure;
+ }
+ return CompletableFuture.completedFuture(response);
+ },
+ (callbackResult, failure) -> {
+ callbackResponse.set(callbackResult);
+ callbackLatch.countDown();
+ },
+ EPOCH,
+ System.currentTimeMillis()));
+
+ assertThat(callbackLatch.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(callbackResponse.get()).isSameAs(response);
+ assertThat(sendCount.get()).isEqualTo(2);
+ assertThat(invalidationCount.get()).isEqualTo(2);
+ }
+
+ @Test
+ void testConnectionInvalidationTimeoutDoesNotResendRequest() throws Exception {
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+ MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup();
+ TabletServerGateway gateway = unusedGateway();
+ CompletableFuture<ApiVersionsResponse> firstAttempt = new CompletableFuture<>();
+ CompletableFuture<Void> invalidationFuture = new CompletableFuture<>();
+ ApiVersionsResponse response = new ApiVersionsResponse();
+ AtomicInteger sendCount = new AtomicInteger();
+ AtomicInteger invalidationCount = new AtomicInteger();
+ CountDownLatch invalidationStarted = new CountDownLatch(1);
+ CountDownLatch callbackLatch = new CountDownLatch(1);
+
+ thread =
+ createThread(
+ queue,
+ () -> Optional.of(gateway),
+ () -> {
+ invalidationCount.incrementAndGet();
+ invalidationStarted.countDown();
+ return invalidationFuture;
+ },
+ () -> EPOCH,
+ metricGroup,
+ Duration.ofMillis(20));
+ thread.start();
+
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored ->
+ sendCount.incrementAndGet() == 1
+ ? firstAttempt
+ : CompletableFuture.completedFuture(response),
+ (callbackResponse, failure) -> callbackLatch.countDown(),
+ EPOCH,
+ System.currentTimeMillis()));
+
+ assertThat(invalidationStarted.await(5, TimeUnit.SECONDS)).isTrue();
+ waitUntil(
+ () -> getCounter(metricGroup, MetricNames.SENDER_RETRY_COUNT).getCount() > 0,
+ Duration.ofSeconds(5),
+ "The invalidation wait did not time out");
+ assertThat(sendCount.get()).isEqualTo(1);
+ assertThat(invalidationCount.get()).isEqualTo(1);
+ assertThat(callbackLatch.getCount()).isEqualTo(1);
+
+ firstAttempt.completeExceptionally(new NetworkException("connection invalidated"));
+ invalidationFuture.complete(null);
+
+ assertThat(callbackLatch.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(sendCount.get()).isEqualTo(2);
+ }
+
+ @Test
+ void testAsyncRequestErrorIsRethrown() throws Exception {
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+ MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup();
+ AssertionError error = new AssertionError("fatal request error");
+ CompletableFuture<ApiVersionsResponse> failedFuture = new CompletableFuture<>();
+ failedFuture.completeExceptionally(error);
+
+ thread = createThread(queue, () -> Optional.of(unusedGateway()), () -> EPOCH, metricGroup);
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> failedFuture,
+ (response, failure) -> {},
+ EPOCH,
+ System.currentTimeMillis()));
+
+ assertThatThrownBy(thread::doWork).isSameAs(error);
+ }
+
+ @Test
+ void testAsyncConnectionInvalidationErrorIsRethrown() throws Exception {
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+ MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup();
+ CompletableFuture<ApiVersionsResponse> requestFailure = new CompletableFuture<>();
+ requestFailure.completeExceptionally(new NetworkException("transient error"));
+ AssertionError error = new AssertionError("fatal invalidation error");
+ CompletableFuture<Void> invalidationFailure = new CompletableFuture<>();
+ invalidationFailure.completeExceptionally(error);
+
+ thread =
+ createThread(
+ queue,
+ () -> Optional.of(unusedGateway()),
+ () -> invalidationFailure,
+ () -> EPOCH,
+ metricGroup,
+ Duration.ofSeconds(30));
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> requestFailure,
+ (response, failure) -> {},
+ EPOCH,
+ System.currentTimeMillis()));
+
+ assertThatThrownBy(thread::doWork).isSameAs(error);
+ }
+
+ @Test
+ void testStaleEpochDrop() throws Exception {
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+ MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup();
+
+ ApiVersionsResponse response = new ApiVersionsResponse();
+ TabletServerGateway gateway = unusedGateway();
+
+ CountDownLatch sentinelLatch = new CountDownLatch(1);
+
+ thread = createThread(queue, () -> Optional.of(gateway), () -> 5, metricGroup);
+ thread.start();
+
+ // stale item (epoch 1 < current 5)
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> CompletableFuture.completedFuture(response),
+ (resp, err) -> {},
+ 1,
+ System.currentTimeMillis()));
+
+ // sentinel item with current epoch to detect when stale item was processed
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> CompletableFuture.completedFuture(response),
+ (resp, err) -> sentinelLatch.countDown(),
+ 5,
+ System.currentTimeMillis()));
+
+ assertThat(sentinelLatch.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(getCounter(metricGroup, MetricNames.SENDER_STALE_DROP_COUNT).getCount())
+ .isEqualTo(1);
+ }
+
+ @Test
+ void testGatewayAbsentThenPresent() throws Exception {
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+ MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup();
+
+ ApiVersionsResponse response = new ApiVersionsResponse();
+ TabletServerGateway gateway = unusedGateway();
+
+ AtomicInteger gatewayCallCount = new AtomicInteger(0);
+ Supplier<Optional<TabletServerGateway>> gatewaySupplier =
+ () -> {
+ if (gatewayCallCount.incrementAndGet() <= 3) {
+ return Optional.empty();
+ }
+ return Optional.of(gateway);
+ };
+
+ CountDownLatch callbackLatch = new CountDownLatch(1);
+
+ thread = createThread(queue, gatewaySupplier, () -> EPOCH, metricGroup);
+ thread.start();
+
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> CompletableFuture.completedFuture(response),
+ (resp, err) -> callbackLatch.countDown(),
+ EPOCH,
+ System.currentTimeMillis()));
+
+ assertThat(callbackLatch.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(getCounter(metricGroup, MetricNames.SENDER_RETRY_COUNT).getCount())
+ .isGreaterThanOrEqualTo(3);
+ }
+
+ @Test
+ void testShutdownCancelsInFlight() throws Exception {
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+ MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup();
+
+ CountDownLatch invokedLatch = new CountDownLatch(1);
+ CompletableFuture<ApiVersionsResponse> neverCompleteFuture = new CompletableFuture<>();
+ TabletServerGateway gateway = unusedGateway();
+
+ AtomicInteger callbackCount = new AtomicInteger(0);
+
+ thread = createThread(queue, () -> Optional.of(gateway), () -> EPOCH, metricGroup);
+ thread.start();
+
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> {
+ invokedLatch.countDown();
+ return neverCompleteFuture;
+ },
+ (resp, err) -> callbackCount.incrementAndGet(),
+ EPOCH,
+ System.currentTimeMillis()));
+
+ assertThat(invokedLatch.await(5, TimeUnit.SECONDS)).isTrue();
+
+ long shutdownStart = System.currentTimeMillis();
+ thread.initiateShutdown();
+ thread.awaitShutdown();
+ long shutdownDuration = System.currentTimeMillis() - shutdownStart;
+
+ assertThat(shutdownDuration).isLessThan(2000);
+ assertThat(callbackCount.get()).isEqualTo(0);
+ assertThat(getGaugeValue(metricGroup, MetricNames.SENDER_ALIVE)).isEqualTo(0);
+ thread = null;
+ }
+
+ @Test
+ void testCallbackThrowsDoesNotRetry() throws Exception {
+ BlockingQueue<QueueItem<?>> queue = new LinkedBlockingQueue<>();
+ MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup();
+
+ ApiVersionsResponse response = new ApiVersionsResponse();
+ AtomicInteger invocationCount = new AtomicInteger(0);
+ TabletServerGateway gateway = unusedGateway();
+
+ CountDownLatch secondCallbackLatch = new CountDownLatch(1);
+
+ thread = createThread(queue, () -> Optional.of(gateway), () -> EPOCH, metricGroup);
+ thread.start();
+
+ // first item: callback throws
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> {
+ invocationCount.incrementAndGet();
+ return CompletableFuture.completedFuture(response);
+ },
+ (resp, err) -> {
+ throw new RuntimeException("buggy callback");
+ },
+ EPOCH,
+ System.currentTimeMillis()));
+
+ // second item: proves thread survived
+ queue.put(
+ new QueueItem<>(
+ ApiKeys.API_VERSIONS,
+ ignored -> {
+ invocationCount.incrementAndGet();
+ return CompletableFuture.completedFuture(response);
+ },
+ (resp, err) -> secondCallbackLatch.countDown(),
+ EPOCH,
+ System.currentTimeMillis()));
+
+ assertThat(secondCallbackLatch.await(5, TimeUnit.SECONDS)).isTrue();
+ // request sender invoked exactly twice (once per item, no re-send from callback failure)
+ assertThat(invocationCount.get()).isEqualTo(2);
+ assertThat(getCounter(metricGroup, MetricNames.SENDER_RETRY_COUNT).getCount()).isEqualTo(0);
+ assertThat(getGaugeValue(metricGroup, MetricNames.SENDER_ALIVE)).isEqualTo(1);
+ }
+
+ private static TabletServerGateway unusedGateway() {
+ return (TabletServerGateway)
+ Proxy.newProxyInstance(
+ TabletServerGateway.class.getClassLoader(),
+ new Class<?>[] {TabletServerGateway.class},
+ (proxy, method, args) -> {
+ throw new UnsupportedOperationException(
+ "Unexpected gateway call: " + method.getName());
+ });
+ }
+
+ private ControlRequestSendThread createThread(
+ BlockingQueue<QueueItem<?>> queue,
+ Supplier<Optional<TabletServerGateway>> gatewaySupplier,
+ IntSupplier epochSupplier,
+ MetricGroup metricGroup) {
+ return createThread(
+ queue,
+ gatewaySupplier,
+ () -> CompletableFuture.completedFuture(null),
+ epochSupplier,
+ metricGroup,
+ Duration.ofSeconds(30));
+ }
+
+ private ControlRequestSendThread createThread(
+ BlockingQueue<QueueItem<?>> queue,
+ Supplier<Optional<TabletServerGateway>> gatewaySupplier,
+ Supplier<CompletableFuture<Void>> connectionInvalidator,
+ IntSupplier epochSupplier,
+ MetricGroup metricGroup,
+ Duration requestTimeout) {
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_CONTROL_REQUEST_RETRY_BACKOFF, Duration.ofMillis(10));
+ conf.set(ConfigOptions.COORDINATOR_CONTROL_REQUEST_TIMEOUT, requestTimeout);
+ return new ControlRequestSendThread(
+ TABLET_SERVER_ID,
+ queue,
+ gatewaySupplier,
+ connectionInvalidator,
+ epochSupplier,
+ conf,
+ metricGroup);
+ }
+
+ private static Counter getCounter(MetricGroup group, String name) {
+ return (Counter) ((TestMetricGroup) group).getMetric(name);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static int getGaugeValue(MetricGroup group, String name) {
+ return ((Gauge<Integer>) ((TestMetricGroup) group).getMetric(name)).getValue();
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachineTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachineTest.java
index 4d1946ab4..94eeb3475 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachineTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachineTest.java
@@ -33,6 +33,7 @@ import org.apache.fluss.server.coordinator.TestCoordinatorChannelManager;
import org.apache.fluss.server.coordinator.event.DeleteReplicaResponseReceivedEvent;
import org.apache.fluss.server.entity.DeleteReplicaResultForBucket;
import org.apache.fluss.server.metadata.ServerInfo;
+import org.apache.fluss.server.metrics.group.TestingMetricGroups;
import org.apache.fluss.server.zk.NOPErrorHandler;
import org.apache.fluss.server.zk.ZkEpoch;
import org.apache.fluss.server.zk.ZooKeeperClient;
@@ -318,7 +319,10 @@ class ReplicaStateMachineTest {
new CoordinatorChannelManager(
RpcClient.create(
new Configuration(),
- TestingClientMetricGroup.newInstance())),
+ TestingClientMetricGroup.newInstance()),
+ () -> 0,
+ new Configuration(),
+ TestingMetricGroups.COORDINATOR_METRICS),
(event) -> {
// do nothing
},
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
index 618512675..feed78946 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
@@ -330,7 +330,10 @@ class TableBucketStateMachineTest {
new CoordinatorChannelManager(
RpcClient.create(
new Configuration(),
- TestingClientMetricGroup.newInstance())),
+ TestingClientMetricGroup.newInstance()),
+ () -> 0,
+ new Configuration(),
+ TestingMetricGroups.COORDINATOR_METRICS),
coordinatorContext,
replicaCapacityController,
autoPartitionManager,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/RpcGatewayManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/RpcGatewayManagerTest.java
index 8e07921a7..24d0fdd61 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/utils/RpcGatewayManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/RpcGatewayManagerTest.java
@@ -17,15 +17,32 @@
package org.apache.fluss.server.utils;
+import org.apache.fluss.cluster.Endpoint;
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.metrics.groups.MetricGroup;
+import org.apache.fluss.metrics.util.NOPMetricsGroup;
+import org.apache.fluss.rpc.TestingTabletGatewayService;
import org.apache.fluss.rpc.gateway.TabletServerGateway;
+import org.apache.fluss.rpc.messages.LookupRequest;
+import org.apache.fluss.rpc.messages.LookupResponse;
import org.apache.fluss.rpc.metrics.TestingClientMetricGroup;
import org.apache.fluss.rpc.netty.client.NettyClient;
+import org.apache.fluss.rpc.netty.server.NettyServer;
+import org.apache.fluss.rpc.netty.server.RequestsMetrics;
+import org.apache.fluss.utils.NetUtils;
import org.junit.jupiter.api.Test;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.fluss.utils.NetUtils.getAvailablePort;
import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link org.apache.fluss.server.utils.RpcGatewayManager}. */
@@ -47,6 +64,10 @@ class RpcGatewayManagerTest {
// shouldn't be empty then
assertThat(gatewayRpcGatewayManager.getRpcGateway(serverNode1.id())).isPresent();
+ // disconnect should retain the gateway so the next request can reconnect
+ gatewayRpcGatewayManager.disconnectServer(serverNode1.id()).get();
+ assertThat(gatewayRpcGatewayManager.getRpcGateway(serverNode1.id())).isPresent();
+
// add the server again
gatewayRpcGatewayManager.addServer(serverNode1);
assertThat(gatewayRpcGatewayManager.getRpcGateway(serverNode1.id())).isPresent();
@@ -67,4 +88,84 @@ class RpcGatewayManagerTest {
gatewayRpcGatewayManager.close();
}
+
+ @Test
+ void testDisconnectFailsInflightRequestAndAllowsReconnect() throws Exception {
+ Configuration conf = new Configuration();
+ conf.setInt(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS, 3);
+ DelayedLookupGatewayService service = new DelayedLookupGatewayService();
+ MetricGroup metricGroup = NOPMetricsGroup.newInstance();
+
+ try (NetUtils.Port availablePort = getAvailablePort()) {
+ ServerNode serverNode =
+ new ServerNode(
+ 1, "localhost", availablePort.getPort(), ServerType.TABLET_SERVER);
+ try (NettyServer nettyServer =
+ new NettyServer(
+ conf,
+ Collections.singleton(
+ new Endpoint(
+ serverNode.host(),
+ serverNode.port(),
+ "INTERNAL")),
+ service,
+ metricGroup,
+ RequestsMetrics.createTabletServerRequestMetrics(metricGroup));
+ NettyClient nettyClient =
+ new NettyClient(conf, TestingClientMetricGroup.newInstance())) {
+ try {
+ nettyServer.start();
+
+ RpcGatewayManager<TabletServerGateway> gatewayManager =
+ new RpcGatewayManager<>(nettyClient, TabletServerGateway.class);
+ gatewayManager.addServer(serverNode);
+ TabletServerGateway gateway =
+ gatewayManager.getRpcGateway(serverNode.id()).get();
+ LookupRequest request = new LookupRequest().setTableId(1);
+ request.addBucketsReq().setBucketId(1);
+
+ CompletableFuture<LookupResponse> firstRequest = gateway.lookup(request);
+ assertThat(service.awaitFirstLookup()).isTrue();
+ assertThat(firstRequest).isNotDone();
+
+ gatewayManager.disconnectServer(serverNode.id()).get(5, TimeUnit.SECONDS);
+ assertThat(firstRequest).isCompletedExceptionally();
+ assertThat(gatewayManager.getRpcGateway(serverNode.id())).containsSame(gateway);
+
+ assertThat(gateway.lookup(request).get(5, TimeUnit.SECONDS)).isNotNull();
+ assertThat(service.getInvocationCount()).isEqualTo(2);
+ } finally {
+ service.completeFirstLookup();
+ }
+ }
+ }
+ }
+
+ private static class DelayedLookupGatewayService extends TestingTabletGatewayService {
+ private final AtomicInteger invocationCount = new AtomicInteger();
+ private final CountDownLatch firstLookup = new CountDownLatch(1);
+ private final CompletableFuture<LookupResponse> firstLookupResponse =
+ new CompletableFuture<>();
+
+ @Override
+ public CompletableFuture<LookupResponse> lookup(LookupRequest request) {
+ if (invocationCount.incrementAndGet() == 1) {
+ firstLookup.countDown();
+ return firstLookupResponse;
+ }
+ return CompletableFuture.completedFuture(new LookupResponse());
+ }
+
+ private boolean awaitFirstLookup() throws InterruptedException {
+ return firstLookup.await(5, TimeUnit.SECONDS);
+ }
+
+ private int getInvocationCount() {
+ return invocationCount.get();
+ }
+
+ private void completeFirstLookup() {
+ firstLookupResponse.complete(new LookupResponse());
+ }
+ }
}
diff --git a/website/docs/maintenance/configuration.md b/website/docs/maintenance/configuration.md
index af5bcb973..49d29db7f 100644
--- a/website/docs/maintenance/configuration.md
+++ b/website/docs/maintenance/configuration.md
@@ -78,6 +78,8 @@ The logging-related environment options (`env.log.dir`, `env.log.level`, `env.lo
| coordinator.io-pool.size | Integer | 10 | **Deprecated**: This option is deprecated. Please use `server.io-pool.size` instead. The size of the IO thread pool to run blocking operations for coordinator server. This includes discard unnecessary snapshot files. Increase this value if you experience slow unnecessary snapshot files clean. The default value is 10. [...]
| coordinator.producer-offsets.ttl | Duration | 24h | The TTL (time-to-live) for producer offsets. Producer offsets older than this TTL will be automatically cleaned up by the coordinator server. Producer offsets are used for undo recovery when a Flink job fails over before completing its first checkpoint. The default value is 24 hours. [...]
| coordinator.producer-offsets.cleanup-interval | Duration | 1h | The interval for cleaning up expired producer offsets and orphan files in remote storage. The cleanup task runs periodically to remove expired offsets and any orphan files that may have been left behind due to incomplete operations. The default value is 1 hour. [...]
+| coordinator.control-request.retry-backoff | Duration | 100ms | The backoff duration the coordinator waits before retrying a control-plane request to a tablet server after a transient RPC-layer failure. [...]
+| coordinator.control-request.timeout | Duration | 30s | The timeout the sender thread waits for a response to a control-plane request before treating it as failed and retrying. [...]
| coordinator.lifecycle-throttler.inflight-timeout | Duration | 3min | The timeout for an in-flight drop event in the coordinator's TableLifecycleThrottler. If a drop event has been admitted but the corresponding completion callback has not arrived within this timeout, the throttler abandons tracking of that drop and continues admitting the next pending drop. [...]
| coordinator.lifecycle-throttler.timeout-check-interval | Duration | 1min | The periodic interval at which the coordinator's TableLifecycleThrottler scans in-flight drops for timeouts. [...]
| coordinator.offline-leader.retry-delay | Duration | 1min | The delay before the coordinator retries offline leaders on live tablet servers after they are marked offline. This lets a leader that was rejected because of temporary tablet-server conditions, such as disk write protection, become electable again after recovery. [...]