From 4b8a703190cf763cf5e282c9372dee38a6fd733b Mon Sep 17 00:00:00 2001 From: litiliu <38579068+litiliu@users.noreply.github.com> Date: Tue, 18 Aug 2026 16:12:25 +0800 Subject: [PATCH] [client] Recover admin writes after coordinator leader failover A long-lived Admin client caches the coordinator leader. After failover to a standby, coordinator write operations (dropDatabase, etc.) kept failing with NotCoordinatorLeaderException because: 1. FlussAdmin wrapped only the read-only gateway with retry, so the write gateway never refreshed metadata after a failover. 2. Even after refreshing to the new leader's address, NettyClient reused the stale connection cached under the coordinator uid "cs-0" (both coordinators share id 0), so requests kept hitting the old leader that is still alive as a standby. Fixes: - RetryableGatewayClientProxy now takes separate refresh and retry predicates. The write gateway refreshes metadata on any recoverable error (NotCoordinatorLeaderException or network errors after a failover/upgrade) so the stale coordinator connection is repointed and a manual retry can recover, but auto-retries only NotCoordinatorLeaderException -- which the server rejects before invoking the write API, so a retry cannot duplicate an already-executed, non-idempotent mutation. Read-only gateways keep retrying any RetriableException. - NettyClient recreates the connection when the address for a server uid changes, closing the stale connection. Tests: - RetryableGatewayClientProxyTest: retry on the safe error; refresh-but-no-retry on network errors; no refresh/retry when neither predicate matches. - NettyClientTest: reconnect when a uid's address changes. - CoordinatorFailoverAdminITCase: keeps one Admin open across a coordinator leader failover and verifies a write succeeds afterward. Closes #4027 --- .../apache/fluss/client/admin/FlussAdmin.java | 28 +- .../admin/CoordinatorFailoverAdminITCase.java | 288 ++++++++++++++++++ .../rpc/RetryableGatewayClientProxy.java | 105 ++++++- .../fluss/rpc/netty/client/NettyClient.java | 55 +++- .../rpc/RetryableGatewayClientProxyTest.java | 125 ++++++++ .../rpc/netty/client/NettyClientTest.java | 52 ++++ 6 files changed, 621 insertions(+), 32 deletions(-) create mode 100644 fluss-client/src/test/java/org/apache/fluss/client/admin/CoordinatorFailoverAdminITCase.java diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java index a0909ceec73..51dc55bd0f8 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java @@ -34,6 +34,8 @@ import org.apache.fluss.config.cluster.ConfigEntry; import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.exception.LeaderNotAvailableException; +import org.apache.fluss.exception.NotCoordinatorLeaderException; +import org.apache.fluss.exception.RetriableException; import org.apache.fluss.metadata.DatabaseChange; import org.apache.fluss.metadata.DatabaseDescriptor; import org.apache.fluss.metadata.DatabaseInfo; @@ -157,15 +159,27 @@ public class FlussAdmin implements Admin { 1, new ExecutorThreadFactory("fluss-admin-metadata-refresh")); public FlussAdmin(RpcClient client, MetadataUpdater metadataUpdater) { - // TODO: AdminGateway includes non-idempotent write operations (createTable, dropTable, - // createDatabase, etc.). Wrapping it with RetryableGatewayClientProxy is unsafe because - // a request may succeed on the server while the response is lost (surfacing as a - // RetriableException), causing a duplicate mutation on retry. A future phase should - // introduce idempotent retry semantics (e.g., request-id deduplication) before enabling - // retry on the write gateway. - this.gateway = + // The write gateway carries non-idempotent operations, so we must NOT auto-retry generic + // RetriableException (a lost response could duplicate an executed mutation). We still + // refresh metadata on any recoverable error (NotCoordinatorLeaderException or network + // errors after a failover/upgrade) so the stale coordinator connection is repointed and a + // manual retry can recover. We only auto-retry NotCoordinatorLeaderException: the + // coordinator + // rejects such a request before invoking the write API (see FlussRequestHandler), so a + // retry cannot duplicate an already-executed mutation. + AdminGateway rawGateway = GatewayClientProxy.createGatewayProxy( metadataUpdater::getCoordinatorServer, client, AdminGateway.class); + this.gateway = + RetryableGatewayClientProxy.createRetryableGatewayProxy( + rawGateway, + metadataUpdater::refreshClusterUntilAvailable, + refreshExecutor, + cause -> + cause instanceof NotCoordinatorLeaderException + || cause instanceof RetriableException, + cause -> cause instanceof NotCoordinatorLeaderException, + AdminGateway.class); AdminGateway rawReadOnlyGateway = GatewayClientProxy.createGatewayProxy( metadataUpdater::getRandomTabletServer, client, AdminGateway.class); diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/CoordinatorFailoverAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/CoordinatorFailoverAdminITCase.java new file mode 100644 index 00000000000..d9bfd25f328 --- /dev/null +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/CoordinatorFailoverAdminITCase.java @@ -0,0 +1,288 @@ +/* + * 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.client.admin; + +import org.apache.fluss.client.Connection; +import org.apache.fluss.client.ConnectionFactory; +import org.apache.fluss.cluster.Endpoint; +import org.apache.fluss.cluster.ServerNode; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.DatabaseDescriptor; +import org.apache.fluss.server.coordinator.CoordinatorServer; +import org.apache.fluss.server.tablet.TabletServer; +import org.apache.fluss.server.zk.NOPErrorHandler; +import org.apache.fluss.server.zk.ZooKeeperClient; +import org.apache.fluss.server.zk.ZooKeeperExtension; +import org.apache.fluss.shaded.curator5.org.apache.curator.framework.CuratorFramework; +import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.Watcher; +import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.ZooKeeper; +import org.apache.fluss.testutils.common.AllCallbackWrapper; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Integration test for admin write recovery after a coordinator leader failover (issue #4027). + * + *

A single long-lived {@link Admin} client caches the coordinator leader in its metadata. When + * leadership moves to the standby while the old leader stays alive, the old leader answers + * coordinator write RPCs with {@code NotCoordinatorLeaderException}. This test verifies the client + * recognizes that error, refreshes metadata, resolves the new leader, and the write succeeds + * without recreating the connection. + */ +class CoordinatorFailoverAdminITCase { + + private static final String CLIENT_LISTENER_NAME = "CLIENT"; + + @RegisterExtension + public static final AllCallbackWrapper ZOO_KEEPER_EXTENSION_WRAPPER = + new AllCallbackWrapper<>(new ZooKeeperExtension()); + + private static ZooKeeperClient zookeeperClient; + + private CoordinatorServer coordinatorServer1; + private CoordinatorServer coordinatorServer2; + private TabletServer tabletServer; + + @TempDir Path tempDir; + + @BeforeAll + static void baseBeforeAll() { + zookeeperClient = + ZOO_KEEPER_EXTENSION_WRAPPER + .getCustomExtension() + .getZooKeeperClient(NOPErrorHandler.INSTANCE); + } + + @AfterEach + void tearDown() throws Exception { + if (tabletServer != null) { + tabletServer.close(); + tabletServer = null; + } + if (coordinatorServer1 != null) { + coordinatorServer1.close(); + coordinatorServer1 = null; + } + if (coordinatorServer2 != null) { + coordinatorServer2.close(); + coordinatorServer2 = null; + } + } + + @Test + void testAdminWriteRecoversAfterCoordinatorFailover() throws Exception { + coordinatorServer1 = new CoordinatorServer(createCoordinatorConfiguration()); + coordinatorServer2 = new CoordinatorServer(createCoordinatorConfiguration()); + tabletServer = new TabletServer(createTabletServerConfiguration()); + + coordinatorServer1.start(); + coordinatorServer2.start(); + tabletServer.start(); + + waitUntilCoordinatorServerElected(); + + CoordinatorServer leader = findLeader(); + CoordinatorServer standby = findStandby(leader); + assertThat(leader).isNotNull(); + assertThat(standby).isNotNull(); + + // A single long-lived connection/admin, bootstrapped against both coordinators. + String db = "test_failover_db"; + try (Connection connection = + ConnectionFactory.createConnection(createClientConfiguration()); + Admin admin = connection.getAdmin()) { + // Initialize the client metadata and cache the current coordinator leader. + admin.createDatabase(db, DatabaseDescriptor.EMPTY, false).get(); + assertThat(admin.databaseExists(db).get()).isTrue(); + + // Trigger failover: kill the leader's ZK session. The old leader process stays alive + // and becomes a standby, so it will reject coordinator writes with + // NotCoordinatorLeaderException instead of dropping the connection. + killZkSession(leader); + waitUntilNewLeaderElected(leader.getServerId()); + assertThat(zookeeperClient.getCoordinatorLeaderAddress().get().getId()) + .as("standby should become the new leader after failover") + .isEqualTo(standby.getServerId()); + // Wait until the tablet server's metadata cache (which feeds the client's metadata + // refresh) reflects the new coordinator leader. + waitUntilTabletServerSeesCoordinator(standby); + + // The same admin still points at the stale coordinator. The write must recover: it hits + // NotCoordinatorLeaderException, refreshes metadata, resolves the new leader, and + // retries. + admin.dropDatabase(db, false, true).get(); + assertThat(admin.databaseExists(db).get()).isFalse(); + } + } + + private void waitUntilTabletServerSeesCoordinator(CoordinatorServer expectedCoordinator) { + Endpoint expected = clientEndpoint(expectedCoordinator); + waitUntil( + () -> { + ServerNode coordinator = + tabletServer + .getMetadataCache() + .getCoordinatorServer(CLIENT_LISTENER_NAME); + return coordinator != null + && coordinator.host().equals(expected.getHost()) + && coordinator.port() == expected.getPort(); + }, + Duration.ofSeconds(30), + "Tablet server did not learn the new coordinator after failover"); + } + + private CoordinatorServer findLeader() throws Exception { + String leaderId = zookeeperClient.getCoordinatorLeaderAddress().get().getId(); + return Objects.equals(coordinatorServer1.getServerId(), leaderId) + ? coordinatorServer1 + : coordinatorServer2; + } + + private CoordinatorServer findStandby(CoordinatorServer leader) { + return leader == coordinatorServer1 ? coordinatorServer2 : coordinatorServer1; + } + + private void waitUntilCoordinatorServerElected() throws Exception { + waitUntil( + () -> zookeeperClient.getCoordinatorLeaderAddress().isPresent(), + Duration.ofMinutes(1), + "Fail to wait for coordinator server to be elected"); + waitUntilCoordinatorLeaderReady(); + } + + private void waitUntilCoordinatorLeaderReady() throws Exception { + CoordinatorServer leader = findLeader(); + waitUntil( + () -> leader.getCoordinatorService().isLeader(), + Duration.ofSeconds(30), + "Coordinator leader did not recognize itself as leader"); + } + + private void waitUntilNewLeaderElected(String oldLeaderId) throws Exception { + waitUntil( + () -> { + try { + return zookeeperClient + .getCoordinatorLeaderAddress() + .map(addr -> !addr.getId().equals(oldLeaderId)) + .orElse(false); + } catch (Exception e) { + return false; + } + }, + Duration.ofMinutes(1), + "Fail to wait for new coordinator leader to be elected"); + waitUntilCoordinatorLeaderReady(); + } + + /** + * Kills the ZK session of a CoordinatorServer to simulate a real session timeout, forcing it to + * lose leadership while the process stays alive as a standby. + */ + private void killZkSession(CoordinatorServer server) throws Exception { + CuratorFramework curatorClient = server.getZooKeeperClient().getCuratorClient(); + ZooKeeper zk = curatorClient.getZookeeperClient().getZooKeeper(); + long sessionId = zk.getSessionId(); + byte[] sessionPasswd = zk.getSessionPasswd(); + String connectString = ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().getConnectString(); + + // Wait for the duplicate connection to be fully established before closing it, otherwise + // the + // ZK server may never see the duplicate session and the original session stays alive. + CountDownLatch connectedLatch = new CountDownLatch(1); + ZooKeeper dupZk = + new ZooKeeper( + connectString, + 1000, + event -> { + if (event.getState() == Watcher.Event.KeeperState.SyncConnected) { + connectedLatch.countDown(); + } + }, + sessionId, + sessionPasswd); + if (!connectedLatch.await(10, TimeUnit.SECONDS)) { + dupZk.close(); + throw new RuntimeException( + "Failed to establish duplicate ZK connection for session kill"); + } + dupZk.close(); + } + + private Configuration createClientConfiguration() { + Configuration conf = new Configuration(); + conf.set( + ConfigOptions.BOOTSTRAP_SERVERS, + Arrays.asList( + clientBootstrap(coordinatorServer1), clientBootstrap(coordinatorServer2))); + return conf; + } + + private static String clientBootstrap(CoordinatorServer server) { + Endpoint endpoint = clientEndpoint(server); + return endpoint.getHost() + ":" + endpoint.getPort(); + } + + private static Endpoint clientEndpoint(CoordinatorServer server) { + List endpoints = server.getRpcServer().getBindEndpoints(); + return endpoints.stream() + .filter(e -> e.getListenerName().equals(CLIENT_LISTENER_NAME)) + .findFirst() + .orElse(endpoints.get(0)); + } + + private Configuration createCoordinatorConfiguration() { + Configuration configuration = new Configuration(); + configuration.setString( + ConfigOptions.ZOOKEEPER_ADDRESS, + ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().getConnectString()); + configuration.setString( + ConfigOptions.BIND_LISTENERS, "CLIENT://localhost:0,FLUSS://localhost:0"); + configuration.setString(ConfigOptions.INTERNAL_LISTENER_NAME, "FLUSS"); + configuration.set(ConfigOptions.REMOTE_DATA_DIR, tempDir.resolve("remote-data").toString()); + // Use a shorter session timeout so the killed leader loses leadership quickly. + configuration.set(ConfigOptions.ZOOKEEPER_SESSION_TIMEOUT, Duration.ofSeconds(5)); + configuration.set(ConfigOptions.ZOOKEEPER_CONNECTION_TIMEOUT, Duration.ofSeconds(5)); + configuration.set(ConfigOptions.ZOOKEEPER_RETRY_WAIT, Duration.ofMillis(500)); + return configuration; + } + + private Configuration createTabletServerConfiguration() { + Configuration configuration = createCoordinatorConfiguration(); + configuration.set(ConfigOptions.TABLET_SERVER_ID, 0); + configuration.setString( + ConfigOptions.DATA_DIR, tempDir.resolve("tablet-data").toAbsolutePath().toString()); + return configuration; + } +} diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/RetryableGatewayClientProxy.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/RetryableGatewayClientProxy.java index cc368e05e12..3c380f348ad 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/RetryableGatewayClientProxy.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/RetryableGatewayClientProxy.java @@ -31,15 +31,23 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Predicate; /** - * A proxy that wraps an existing {@link RpcGateway} proxy and adds automatic retry with metadata - * refresh on retriable (network) errors. + * A proxy that wraps an existing {@link RpcGateway} proxy and adds automatic metadata refresh on + * errors, plus an optional single retry of a subset of them. * *

This is designed to solve the stale metadata problem where cached server addresses become - * invalid (e.g., during rolling upgrades in Kubernetes). When an RPC call fails with a {@link - * RetriableException}, this proxy triggers a metadata refresh callback and retries the request with - * potentially updated server addresses. + * invalid (e.g., during rolling upgrades in Kubernetes) or a cached coordinator leader becomes a + * standby after failover. When an RPC fails with an error accepted by {@code refreshPredicate}, + * this proxy triggers a metadata refresh callback; when the error is also accepted by {@code + * retryPredicate}, it retries the request once with the potentially updated server addresses. + * + *

Separating the two predicates lets a write gateway refresh metadata on any recoverable error + * (so the connection is repointed and a later manual retry can succeed) while only auto-retrying + * failures that are provably safe to replay -- e.g. {@code NotCoordinatorLeaderException}, which + * the server raises before executing a non-idempotent mutation. Read-only gateways use {@link + * RetriableException} for both. * *

The retry flow for a cluster with stale tablet servers: * @@ -68,6 +76,8 @@ public class RetryableGatewayClientProxy implements InvocationHandler { private final Object delegate; private final Runnable metadataRefreshAction; private final Executor refreshExecutor; + private final Predicate refreshPredicate; + private final Predicate retryPredicate; /** * Holds the currently in-flight metadata refresh, if any. Concurrent retriers piggyback on this @@ -78,21 +88,66 @@ public class RetryableGatewayClientProxy implements InvocationHandler { new AtomicReference<>(); RetryableGatewayClientProxy( - Object delegate, Runnable metadataRefreshAction, Executor refreshExecutor) { + Object delegate, + Runnable metadataRefreshAction, + Executor refreshExecutor, + Predicate refreshPredicate, + Predicate retryPredicate) { this.delegate = delegate; this.metadataRefreshAction = metadataRefreshAction; this.refreshExecutor = refreshExecutor; + this.refreshPredicate = refreshPredicate; + this.retryPredicate = retryPredicate; } /** * Creates a retryable proxy wrapping an existing gateway proxy. On {@link RetriableException}, - * the proxy will invoke {@code metadataRefreshAction} and retry the failed RPC call once. + * the proxy will invoke {@code metadataRefreshAction} and retry the failed RPC call once. This + * is suitable for read-only gateways where retrying any network error is safe. + * + * @param delegate the underlying gateway proxy to wrap + * @param metadataRefreshAction callback to refresh metadata (e.g., update cluster info) + * @param refreshExecutor executor on which {@code metadataRefreshAction} is run; must NOT be a + * Netty event loop and ideally should be a dedicated, single-thread executor (the in-flight + * refresh is already coalesced to at most one concurrent task) + * @param gatewayClass the gateway interface class + * @param the gateway type + * @return a retryable gateway proxy + */ + public static T createRetryableGatewayProxy( + T delegate, + Runnable metadataRefreshAction, + Executor refreshExecutor, + Class gatewayClass) { + Predicate retriable = cause -> cause instanceof RetriableException; + return createRetryableGatewayProxy( + delegate, + metadataRefreshAction, + refreshExecutor, + retriable, + retriable, + gatewayClass); + } + + /** + * Creates a retryable proxy wrapping an existing gateway proxy. On a failure, the proxy invokes + * {@code metadataRefreshAction} when {@code refreshPredicate} accepts the cause, and then + * retries the RPC once only when {@code retryPredicate} also accepts it. + * + *

This lets a write gateway refresh metadata on any recoverable error (repointing the stale + * coordinator connection so a later manual retry can succeed) while auto-retrying only failures + * that are safe to replay, e.g. {@code NotCoordinatorLeaderException}, which the server raises + * before executing the mutation. Errors such as {@code NetworkException} may already have + * executed on the server, so they refresh metadata but are not auto-retried. * * @param delegate the underlying gateway proxy to wrap * @param metadataRefreshAction callback to refresh metadata (e.g., update cluster info) * @param refreshExecutor executor on which {@code metadataRefreshAction} is run; must NOT be a * Netty event loop and ideally should be a dedicated, single-thread executor (the in-flight * refresh is already coalesced to at most one concurrent task) + * @param refreshPredicate decides whether a failure cause should trigger a metadata refresh + * @param retryPredicate decides whether a failure cause should additionally be retried once + * (should be a subset of {@code refreshPredicate}) * @param gatewayClass the gateway interface class * @param the gateway type * @return a retryable gateway proxy @@ -101,6 +156,8 @@ public static T createRetryableGatewayProxy( T delegate, Runnable metadataRefreshAction, Executor refreshExecutor, + Predicate refreshPredicate, + Predicate retryPredicate, Class gatewayClass) { ClassLoader classLoader = gatewayClass.getClassLoader(); @@ -111,7 +168,11 @@ public static T createRetryableGatewayProxy( classLoader, new Class[] {gatewayClass}, new RetryableGatewayClientProxy( - delegate, metadataRefreshAction, refreshExecutor)); + delegate, + metadataRefreshAction, + refreshExecutor, + refreshPredicate, + retryPredicate)); return proxy; } @@ -143,22 +204,38 @@ private CompletableFuture invokeWithRetry(Method method, Object[] args, b return; } Throwable cause = ExceptionUtils.stripCompletionException(throwable); - if (!(cause instanceof RetriableException) || !retry) { + // Only the initial attempt may refresh/retry; a failed retry gives up. + if (!retry) { + resultFuture.completeExceptionally(cause); + return; + } + boolean shouldRetry = retryPredicate.test(cause); + boolean shouldRefresh = shouldRetry || refreshPredicate.test(cause); + if (!shouldRefresh) { resultFuture.completeExceptionally(cause); return; } LOG.warn( - "RPC call {} failed with retriable error, " - + "refreshing metadata and retrying once.", + "RPC call {} failed, refreshing metadata{}.", method.getName(), + shouldRetry ? " and retrying once" : " (not retrying)", cause); // Coalesce concurrent refreshes so N parallel failing calls trigger only one // metadata refresh (and one round of MetadataUpdater lock contention). coalescedRefresh() .thenCompose( - ignored -> - RetryableGatewayClientProxy.this.invokeWithRetry( - method, args, false)) + ignored -> { + if (shouldRetry) { + return RetryableGatewayClientProxy.this + .invokeWithRetry(method, args, false); + } + // Metadata was refreshed but the failure is not safe to + // auto-retry; surface the original error for a manual + // retry. + CompletableFuture notRetried = new CompletableFuture<>(); + notRetried.completeExceptionally(cause); + return notRetried; + }) .whenComplete( (retryResult, retryError) -> { if (retryError != null) { 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 50e714d0269..86ad77f3fbd 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 @@ -46,6 +46,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; import static org.apache.fluss.utils.Preconditions.checkArgument; @@ -182,17 +183,49 @@ public void close() throws Exception { private 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)); - }); + // A server uid (e.g., the coordinator's "cs-0") can point at a new address after a + // failover: the standby that takes over reuses the same uid but binds a different + // host/port. If we blindly reused the cached connection, requests would keep going to the + // old (now stale) server. So when the cached connection targets a different address, + // replace it and close the stale one. + AtomicReference staleConnection = new AtomicReference<>(); + ServerConnection connection = + connections.compute( + serverId, + (ignored, existing) -> { + if (existing != null && isSameAddress(existing.getServerNode(), node)) { + return existing; + } + if (existing != null) { + LOG.debug( + "Address for server {} changed from {}:{} to {}:{}, recreating connection.", + serverId, + existing.getServerNode().host(), + existing.getServerNode().port(), + node.host(), + node.port()); + staleConnection.set(existing); + } else { + LOG.debug("Creating connection to server {}.", node); + } + return new ServerConnection( + bootstrap, + node, + clientMetricGroup, + authenticatorSupplier.get(), + (con, ignore) -> connections.remove(serverId, con)); + }); + // Close the stale connection outside compute() to avoid re-entrant modification of the map + // from the connection's removal callback; close() is asynchronous. + ServerConnection stale = staleConnection.get(); + if (stale != null) { + stale.close(); + } + return connection; + } + + private static boolean isSameAddress(ServerNode a, ServerNode b) { + return a.port() == b.port() && a.host().equals(b.host()); } @VisibleForTesting diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/RetryableGatewayClientProxyTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/RetryableGatewayClientProxyTest.java index d4c8f9dc382..66f7db04b62 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/RetryableGatewayClientProxyTest.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/RetryableGatewayClientProxyTest.java @@ -18,6 +18,8 @@ package org.apache.fluss.rpc; import org.apache.fluss.exception.NetworkException; +import org.apache.fluss.exception.NotCoordinatorLeaderException; +import org.apache.fluss.exception.RetriableException; import org.apache.fluss.exception.TableNotExistException; import org.apache.fluss.rpc.messages.ApiVersionsRequest; import org.apache.fluss.rpc.messages.ApiVersionsResponse; @@ -232,6 +234,129 @@ void testConcurrentFailingCallsShareSingleRefresh() throws Exception { .isEqualTo(1); } + /** + * Verifies that a caller-supplied predicate retries only on the matching exception. This + * mirrors the write (coordinator) gateway which retries exclusively on {@link + * NotCoordinatorLeaderException} after a coordinator leader failover. + */ + @Test + void testCustomPredicateRetriesOnMatchingExceptionThenSuccess() throws Exception { + AtomicInteger callCount = new AtomicInteger(0); + AtomicInteger refreshCount = new AtomicInteger(0); + + // Fail the first call with NotCoordinatorLeaderException, then succeed on the single retry. + RpcGateway delegate = + new TestRpcGateway() { + @Override + public CompletableFuture apiVersions( + ApiVersionsRequest request) { + int count = callCount.incrementAndGet(); + CompletableFuture future = new CompletableFuture<>(); + if (count == 1) { + future.completeExceptionally( + new NotCoordinatorLeaderException("not the current leader")); + } else { + future.complete(new ApiVersionsResponse()); + } + return future; + } + }; + + RpcGateway proxy = + RetryableGatewayClientProxy.createRetryableGatewayProxy( + delegate, + refreshCount::incrementAndGet, + REFRESH_EXECUTOR, + cause -> + cause instanceof NotCoordinatorLeaderException + || cause instanceof RetriableException, + cause -> cause instanceof NotCoordinatorLeaderException, + RpcGateway.class); + + CompletableFuture result = proxy.apiVersions(new ApiVersionsRequest()); + assertThat(result.get()).isNotNull(); + // Initial call + 1 retry = 2 total calls + assertThat(callCount.get()).isEqualTo(2); + // Metadata refresh should be called once before the retry + assertThat(refreshCount.get()).isEqualTo(1); + } + + /** + * Verifies the write-gateway policy: a network error refreshes metadata (to repoint the stale + * connection) but is NOT auto-retried, because a non-idempotent mutation may already have + * executed on the server. The user can recover by retrying manually. + */ + @Test + void testRefreshesButDoesNotRetryWhenOnlyRefreshPredicateMatches() { + AtomicInteger callCount = new AtomicInteger(0); + AtomicInteger refreshCount = new AtomicInteger(0); + + // Always fail with NetworkException (a RetriableException). + RpcGateway delegate = createGateway(callCount, Integer.MAX_VALUE); + + RpcGateway proxy = + RetryableGatewayClientProxy.createRetryableGatewayProxy( + delegate, + refreshCount::incrementAndGet, + REFRESH_EXECUTOR, + cause -> + cause instanceof NotCoordinatorLeaderException + || cause instanceof RetriableException, + cause -> cause instanceof NotCoordinatorLeaderException, + RpcGateway.class); + + CompletableFuture result = proxy.apiVersions(new ApiVersionsRequest()); + assertThatThrownBy(result::get) + .isInstanceOf(ExecutionException.class) + .rootCause() + .isInstanceOf(NetworkException.class); + // Only the initial call happened (no retry), but metadata was refreshed once. + assertThat(callCount.get()).isEqualTo(1); + assertThat(refreshCount.get()).isEqualTo(1); + } + + /** + * Verifies that a failure matching neither predicate neither refreshes metadata nor retries; + * the original error is surfaced directly. + */ + @Test + void testNoRefreshAndNoRetryWhenNeitherPredicateMatches() { + AtomicInteger callCount = new AtomicInteger(0); + AtomicInteger refreshCount = new AtomicInteger(0); + + RpcGateway delegate = + new TestRpcGateway() { + @Override + public CompletableFuture apiVersions( + ApiVersionsRequest request) { + callCount.incrementAndGet(); + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally( + new TableNotExistException("table does not exist")); + return future; + } + }; + + RpcGateway proxy = + RetryableGatewayClientProxy.createRetryableGatewayProxy( + delegate, + refreshCount::incrementAndGet, + REFRESH_EXECUTOR, + cause -> + cause instanceof NotCoordinatorLeaderException + || cause instanceof RetriableException, + cause -> cause instanceof NotCoordinatorLeaderException, + RpcGateway.class); + + CompletableFuture result = proxy.apiVersions(new ApiVersionsRequest()); + assertThatThrownBy(result::get) + .isInstanceOf(ExecutionException.class) + .rootCause() + .isInstanceOf(TableNotExistException.class); + assertThat(callCount.get()).isEqualTo(1); + assertThat(refreshCount.get()).isEqualTo(0); + } + /** * Creates a test gateway that fails with {@link NetworkException} for the first {@code * failCount} invocations, then returns a successful response. 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 329f5d2f289..f057ff27027 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 @@ -253,6 +253,58 @@ void testExceptionWhenInitializeServerConnection() throws Exception { assertThat(nettyClient.connections()).isEmpty(); } + @Test + void testReconnectWhenServerAddressChangesForSameUid() throws Exception { + MetricGroup metricGroup = NOPMetricsGroup.newInstance(); + ApiVersionsRequest request = + new ApiVersionsRequest() + .setClientSoftwareName("testing_client") + .setClientSoftwareVersion("1.0"); + try (NetUtils.Port port1 = getAvailablePort(); + NetUtils.Port port2 = getAvailablePort(); + NettyServer server1 = + new NettyServer( + conf, + Collections.singleton( + new Endpoint("localhost", port1.getPort(), "INTERNAL")), + new TestingGatewayService(), + metricGroup, + RequestsMetrics.createCoordinatorServerRequestMetrics( + metricGroup)); + NettyServer server2 = + new NettyServer( + conf, + Collections.singleton( + new Endpoint("localhost", port2.getPort(), "INTERNAL")), + new TestingGatewayService(), + metricGroup, + RequestsMetrics.createCoordinatorServerRequestMetrics( + metricGroup))) { + server1.start(); + server2.start(); + + // Both nodes share the same uid "cs-0" but point at different addresses, mimicking a + // coordinator leader failover where the standby takes over the same server id. + ServerNode node1 = + new ServerNode(0, "localhost", port1.getPort(), ServerType.COORDINATOR); + ServerNode node2 = + new ServerNode(0, "localhost", port2.getPort(), ServerType.COORDINATOR); + assertThat(node1.uid()).isEqualTo(node2.uid()); + + nettyClient.sendRequest(node1, ApiKeys.API_VERSIONS, request).get(); + assertThat(nettyClient.connections()).hasSize(1); + assertThat(nettyClient.connections().get(node1.uid()).getServerNode().port()) + .isEqualTo(port1.getPort()); + + // Sending to the same uid at a new address must reconnect to the new address instead of + // reusing the stale connection to the old server. + nettyClient.sendRequest(node2, ApiKeys.API_VERSIONS, request).get(); + assertThat(nettyClient.connections()).hasSize(1); + assertThat(nettyClient.connections().get(node2.uid()).getServerNode().port()) + .isEqualTo(port2.getPort()); + } + } + private void buildNettyServer(int serverId) throws Exception { try (NetUtils.Port availablePort = getAvailablePort()) { serverNode =