From a3c65cdcd43bba3857472d9c042680366d6032fa Mon Sep 17 00:00:00 2001 From: sakshichitnis27 <156598682+sakshichitnis27@users.noreply.github.com> Date: Tue, 1 Sep 2026 09:33:55 +0000 Subject: [PATCH 1/3] [client] Recover Admin writes after coordinator failover --- .../apache/fluss/client/admin/FlussAdmin.java | 30 +++- .../admin/CustomFlussClusterITCase.java | 129 ++++++++++++++++++ .../rpc/RetryableGatewayClientProxy.java | 42 +++++- .../rpc/RetryableGatewayClientProxyTest.java | 24 ++++ 4 files changed, 215 insertions(+), 10 deletions(-) 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..e01744f86b8 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,7 @@ 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.metadata.DatabaseChange; import org.apache.fluss.metadata.DatabaseDescriptor; import org.apache.fluss.metadata.DatabaseInfo; @@ -157,15 +158,19 @@ 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 = + AdminGateway rawGateway = GatewayClientProxy.createGatewayProxy( metadataUpdater::getCoordinatorServer, client, AdminGateway.class); + // Retrying generic network errors is unsafe for non-idempotent writes because the request + // may already have succeeded. NotCoordinatorLeaderException is safe because the standby + // rejects the request before invoking the coordinator API. + this.gateway = + RetryableGatewayClientProxy.createRetryableGatewayProxy( + rawGateway, + () -> refreshCoordinatorMetadata(client, metadataUpdater), + refreshExecutor, + NotCoordinatorLeaderException.class::isInstance, + AdminGateway.class); AdminGateway rawReadOnlyGateway = GatewayClientProxy.createGatewayProxy( metadataUpdater::getRandomTabletServer, client, AdminGateway.class); @@ -178,6 +183,17 @@ public FlussAdmin(RpcClient client, MetadataUpdater metadataUpdater) { this.metadataUpdater = metadataUpdater; } + private static void refreshCoordinatorMetadata( + RpcClient client, MetadataUpdater metadataUpdater) { + metadataUpdater.refreshClusterUntilAvailable(); + ServerNode coordinator = metadataUpdater.getCoordinatorServer(); + if (coordinator != null) { + // Coordinator nodes share the same cs-0 UID. Discard the connection that returned + // NotCoordinatorLeaderException so the retry opens one to the refreshed endpoint. + client.disconnect(coordinator.uid()).join(); + } + } + @Override public CompletableFuture> getServerNodes() { CompletableFuture> future = new CompletableFuture<>(); diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/CustomFlussClusterITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/CustomFlussClusterITCase.java index c5595c5d43d..605be8e79ab 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/CustomFlussClusterITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/CustomFlussClusterITCase.java @@ -26,6 +26,8 @@ import org.apache.fluss.client.table.scanner.log.LogScanner; import org.apache.fluss.client.table.scanner.log.ScanRecords; import org.apache.fluss.client.table.writer.UpsertWriter; +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.config.MemorySize; @@ -37,18 +39,26 @@ import org.apache.fluss.metadata.TablePath; import org.apache.fluss.record.ChangeType; import org.apache.fluss.row.InternalRow; +import org.apache.fluss.server.coordinator.CoordinatorServer; import org.apache.fluss.server.testutils.FlussClusterExtension; import org.apache.fluss.server.zk.ZooKeeperClient; +import org.apache.fluss.server.zk.data.CoordinatorAddress; +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.types.RowType; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import java.nio.file.Path; import java.time.Duration; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -60,11 +70,107 @@ import static org.apache.fluss.testutils.DataTestUtils.row; import static org.apache.fluss.testutils.InternalRowAssert.assertThatRow; import static org.apache.fluss.testutils.common.CommonTestUtils.retry; +import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil; import static org.assertj.core.api.Assertions.assertThat; /** IT case for tests that require manual cluster management. */ class CustomFlussClusterITCase { + @Test + void testAdminWriteRecoversAfterCoordinatorFailover(@TempDir Path tempDir) throws Exception { + final FlussClusterExtension flussClusterExtension = + FlussClusterExtension.builder().setNumOfTabletServers(1).build(); + CoordinatorServer standbyCoordinator = null; + try { + flussClusterExtension.start(); + CoordinatorServer firstLeader = flussClusterExtension.getCoordinatorServer(); + String zooKeeperConnectString = + firstLeader + .getZooKeeperClient() + .getCuratorClient() + .getZookeeperClient() + .getCurrentConnectionString(); + + Configuration standbyConf = new Configuration(); + standbyConf.setString(ConfigOptions.ZOOKEEPER_ADDRESS, zooKeeperConnectString); + standbyConf.setString(ConfigOptions.BIND_LISTENERS, "FLUSS://localhost:0"); + standbyConf.set( + ConfigOptions.REMOTE_DATA_DIR, + tempDir.resolve("standby-remote-data").toString()); + standbyCoordinator = new CoordinatorServer(standbyConf); + standbyCoordinator.start(); + + waitUntil( + () -> + flussClusterExtension + .getZooKeeperClient() + .getCoordinatorServerList() + .size() + == 2, + Duration.ofSeconds(30), + "Standby coordinator did not register"); + + try (Connection connection = + ConnectionFactory.createConnection( + flussClusterExtension.getClientConfig()); + Admin admin = connection.getAdmin()) { + String databaseName = "test_admin_write_after_coordinator_failover"; + admin.createDatabase(databaseName, DatabaseDescriptor.EMPTY, false).get(); + assertThat(admin.listDatabases().get()).contains(databaseName); + + killZooKeeperSession(firstLeader, zooKeeperConnectString); + CoordinatorServer newLeader = standbyCoordinator; + waitUntil( + () -> { + CoordinatorAddress leaderAddress = + flussClusterExtension + .getZooKeeperClient() + .getCoordinatorLeaderAddress() + .orElse(null); + return leaderAddress != null + && leaderAddress.getId().equals(newLeader.getServerId()) + && newLeader.getCoordinatorService().isLeader(); + }, + Duration.ofMinutes(1), + "Standby coordinator did not become leader"); + + Endpoint newLeaderEndpoint = + newLeader.getRpcServer().getBindEndpoints().stream() + .filter( + endpoint -> + endpoint.getListenerName() + .equals( + ConfigOptions.INTERNAL_LISTENER_NAME + .defaultValue())) + .findFirst() + .orElseThrow(IllegalStateException::new); + waitUntil( + () -> { + ServerNode cachedCoordinator = + flussClusterExtension + .getTabletServerById(0) + .getMetadataCache() + .getCoordinatorServer( + ConfigOptions.INTERNAL_LISTENER_NAME + .defaultValue()); + return cachedCoordinator != null + && cachedCoordinator.host().equals(newLeaderEndpoint.getHost()) + && cachedCoordinator.port() == newLeaderEndpoint.getPort(); + }, + Duration.ofSeconds(30), + "Tablet server did not learn the new coordinator leader"); + + admin.dropDatabase(databaseName, false, false).get(); + assertThat(admin.listDatabases().get()).doesNotContain(databaseName); + } + } finally { + if (standbyCoordinator != null) { + standbyCoordinator.close(); + } + flussClusterExtension.close(); + } + } + @Test void testProjectionPushdownWithEmptyBatches() throws Exception { Configuration conf = initConfig(); @@ -326,4 +432,27 @@ protected static Configuration initConfig() { conf.set(ConfigOptions.NETTY_CLIENT_NUM_NETWORK_THREADS, 1); return conf; } + + private static void killZooKeeperSession( + CoordinatorServer server, String zooKeeperConnectString) throws Exception { + CuratorFramework curatorClient = server.getZooKeeperClient().getCuratorClient(); + ZooKeeper zooKeeper = curatorClient.getZookeeperClient().getZooKeeper(); + CountDownLatch connectedLatch = new CountDownLatch(1); + ZooKeeper duplicateSession = + new ZooKeeper( + zooKeeperConnectString, + 1000, + event -> { + if (event.getState() == Watcher.Event.KeeperState.SyncConnected) { + connectedLatch.countDown(); + } + }, + zooKeeper.getSessionId(), + zooKeeper.getSessionPasswd()); + if (!connectedLatch.await(10, TimeUnit.SECONDS)) { + duplicateSession.close(); + throw new IllegalStateException("Failed to connect duplicate ZooKeeper session"); + } + duplicateSession.close(); + } } 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..c017ee27039 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,6 +31,7 @@ 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 @@ -68,6 +69,7 @@ public class RetryableGatewayClientProxy implements InvocationHandler { private final Object delegate; private final Runnable metadataRefreshAction; private final Executor refreshExecutor; + private final Predicate retryPredicate; /** * Holds the currently in-flight metadata refresh, if any. Concurrent retriers piggyback on this @@ -78,10 +80,14 @@ public class RetryableGatewayClientProxy implements InvocationHandler { new AtomicReference<>(); RetryableGatewayClientProxy( - Object delegate, Runnable metadataRefreshAction, Executor refreshExecutor) { + Object delegate, + Runnable metadataRefreshAction, + Executor refreshExecutor, + Predicate retryPredicate) { this.delegate = delegate; this.metadataRefreshAction = metadataRefreshAction; this.refreshExecutor = refreshExecutor; + this.retryPredicate = retryPredicate; } /** @@ -102,6 +108,33 @@ public static T createRetryableGatewayProxy( Runnable metadataRefreshAction, Executor refreshExecutor, Class gatewayClass) { + return createRetryableGatewayProxy( + delegate, + metadataRefreshAction, + refreshExecutor, + RetriableException.class::isInstance, + gatewayClass); + } + + /** + * Creates a retryable proxy wrapping an existing gateway proxy. When an error matches {@code + * retryPredicate}, the proxy will invoke {@code metadataRefreshAction} and retry the failed RPC + * call once. + * + * @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 + * @param retryPredicate predicate that selects errors safe to retry + * @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, + Predicate retryPredicate, + Class gatewayClass) { ClassLoader classLoader = gatewayClass.getClassLoader(); @SuppressWarnings("unchecked") @@ -111,7 +144,10 @@ public static T createRetryableGatewayProxy( classLoader, new Class[] {gatewayClass}, new RetryableGatewayClientProxy( - delegate, metadataRefreshAction, refreshExecutor)); + delegate, + metadataRefreshAction, + refreshExecutor, + retryPredicate)); return proxy; } @@ -143,7 +179,7 @@ private CompletableFuture invokeWithRetry(Method method, Object[] args, b return; } Throwable cause = ExceptionUtils.stripCompletionException(throwable); - if (!(cause instanceof RetriableException) || !retry) { + if (!retry || !retryPredicate.test(cause)) { resultFuture.completeExceptionally(cause); return; } 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..30fbe0d0a91 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,7 @@ package org.apache.fluss.rpc; import org.apache.fluss.exception.NetworkException; +import org.apache.fluss.exception.NotCoordinatorLeaderException; import org.apache.fluss.exception.TableNotExistException; import org.apache.fluss.rpc.messages.ApiVersionsRequest; import org.apache.fluss.rpc.messages.ApiVersionsResponse; @@ -148,6 +149,29 @@ public CompletableFuture apiVersions( assertThat(refreshCount.get()).isEqualTo(0); } + @Test + void testCustomRetryPredicateExcludesNetworkErrors() { + AtomicInteger callCount = new AtomicInteger(0); + AtomicInteger refreshCount = new AtomicInteger(0); + + RpcGateway delegate = createGateway(callCount, 1); + RpcGateway proxy = + RetryableGatewayClientProxy.createRetryableGatewayProxy( + delegate, + refreshCount::incrementAndGet, + REFRESH_EXECUTOR, + NotCoordinatorLeaderException.class::isInstance, + RpcGateway.class); + + CompletableFuture result = proxy.apiVersions(new ApiVersionsRequest()); + assertThatThrownBy(result::get) + .isInstanceOf(ExecutionException.class) + .rootCause() + .isInstanceOf(NetworkException.class); + assertThat(callCount.get()).isEqualTo(1); + assertThat(refreshCount.get()).isEqualTo(0); + } + @Test void testMetadataRefreshFailureDoesNotPreventRetry() throws Exception { AtomicInteger callCount = new AtomicInteger(0); From b66d60a63304fc74e1e12cf31737bfc88df2e2e3 Mon Sep 17 00:00:00 2001 From: sakshichitnis27 <156598682+sakshichitnis27@users.noreply.github.com> Date: Thu, 3 Sep 2026 11:13:30 +0000 Subject: [PATCH 2/3] [client] Separate metadata refresh from write retry --- .../apache/fluss/client/admin/FlussAdmin.java | 10 +++-- .../rpc/RetryableGatewayClientProxy.java | 30 +++++++++---- .../rpc/RetryableGatewayClientProxyTest.java | 42 ++++++++++++++++++- 3 files changed, 69 insertions(+), 13 deletions(-) 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 e01744f86b8..5ed0a8af7c9 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 @@ -35,6 +35,7 @@ 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; @@ -161,14 +162,17 @@ public FlussAdmin(RpcClient client, MetadataUpdater metadataUpdater) { AdminGateway rawGateway = GatewayClientProxy.createGatewayProxy( metadataUpdater::getCoordinatorServer, client, AdminGateway.class); - // Retrying generic network errors is unsafe for non-idempotent writes because the request - // may already have succeeded. NotCoordinatorLeaderException is safe because the standby - // rejects the request before invoking the coordinator API. + // Refresh metadata for recoverable failures, but don't retry generic network errors because + // a non-idempotent write may already have succeeded. NotCoordinatorLeaderException is safe + // to retry because the standby rejects the request before invoking the coordinator API. this.gateway = RetryableGatewayClientProxy.createRetryableGatewayProxy( rawGateway, () -> refreshCoordinatorMetadata(client, metadataUpdater), refreshExecutor, + cause -> + cause instanceof NotCoordinatorLeaderException + || cause instanceof RetriableException, NotCoordinatorLeaderException.class::isInstance, AdminGateway.class); AdminGateway rawReadOnlyGateway = 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 c017ee27039..317ba02c70b 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 @@ -69,6 +69,7 @@ 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; /** @@ -83,10 +84,12 @@ public class RetryableGatewayClientProxy implements InvocationHandler { 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; } @@ -113,17 +116,18 @@ public static T createRetryableGatewayProxy( metadataRefreshAction, refreshExecutor, RetriableException.class::isInstance, + RetriableException.class::isInstance, gatewayClass); } /** - * Creates a retryable proxy wrapping an existing gateway proxy. When an error matches {@code - * retryPredicate}, the proxy will invoke {@code metadataRefreshAction} and retry the failed RPC - * call once. + * Creates a retryable proxy wrapping an existing gateway proxy. Matching errors refresh + * metadata, and errors that also match {@code retryPredicate} retry the failed RPC call once. * * @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 + * @param refreshPredicate predicate that selects errors which require a metadata refresh * @param retryPredicate predicate that selects errors safe to retry * @param gatewayClass the gateway interface class * @param the gateway type @@ -133,6 +137,7 @@ public static T createRetryableGatewayProxy( T delegate, Runnable metadataRefreshAction, Executor refreshExecutor, + Predicate refreshPredicate, Predicate retryPredicate, Class gatewayClass) { ClassLoader classLoader = gatewayClass.getClassLoader(); @@ -147,6 +152,7 @@ public static T createRetryableGatewayProxy( delegate, metadataRefreshAction, refreshExecutor, + refreshPredicate, retryPredicate)); return proxy; } @@ -179,22 +185,30 @@ private CompletableFuture invokeWithRetry(Method method, Object[] args, b return; } Throwable cause = ExceptionUtils.stripCompletionException(throwable); - if (!retry || !retryPredicate.test(cause)) { + 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" : " without 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)) + shouldRetry + ? RetryableGatewayClientProxy.this + .invokeWithRetry(method, args, false) + : future) .whenComplete( (retryResult, retryError) -> { if (retryError != null) { 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 30fbe0d0a91..4c7fdcc06a5 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 @@ -150,7 +150,7 @@ public CompletableFuture apiVersions( } @Test - void testCustomRetryPredicateExcludesNetworkErrors() { + void testCustomPredicatesRefreshWithoutRetryingNetworkError() throws Exception { AtomicInteger callCount = new AtomicInteger(0); AtomicInteger refreshCount = new AtomicInteger(0); @@ -160,6 +160,7 @@ void testCustomRetryPredicateExcludesNetworkErrors() { delegate, refreshCount::incrementAndGet, REFRESH_EXECUTOR, + NetworkException.class::isInstance, NotCoordinatorLeaderException.class::isInstance, RpcGateway.class); @@ -169,7 +170,44 @@ void testCustomRetryPredicateExcludesNetworkErrors() { .rootCause() .isInstanceOf(NetworkException.class); assertThat(callCount.get()).isEqualTo(1); - assertThat(refreshCount.get()).isEqualTo(0); + assertThat(refreshCount.get()).isEqualTo(1); + + assertThat(proxy.apiVersions(new ApiVersionsRequest()).get()).isNotNull(); + assertThat(callCount.get()).isEqualTo(2); + assertThat(refreshCount.get()).isEqualTo(1); + } + + @Test + void testCustomPredicatesRetryNotCoordinatorLeader() throws Exception { + AtomicInteger callCount = new AtomicInteger(0); + AtomicInteger refreshCount = new AtomicInteger(0); + RpcGateway delegate = + new TestRpcGateway() { + @Override + public CompletableFuture apiVersions( + ApiVersionsRequest request) { + if (callCount.incrementAndGet() == 1) { + CompletableFuture failed = + new CompletableFuture<>(); + failed.completeExceptionally( + new NotCoordinatorLeaderException("not coordinator leader")); + return failed; + } + return CompletableFuture.completedFuture(new ApiVersionsResponse()); + } + }; + RpcGateway proxy = + RetryableGatewayClientProxy.createRetryableGatewayProxy( + delegate, + refreshCount::incrementAndGet, + REFRESH_EXECUTOR, + NotCoordinatorLeaderException.class::isInstance, + NotCoordinatorLeaderException.class::isInstance, + RpcGateway.class); + + assertThat(proxy.apiVersions(new ApiVersionsRequest()).get()).isNotNull(); + assertThat(callCount.get()).isEqualTo(2); + assertThat(refreshCount.get()).isEqualTo(1); } @Test From 2844cecbd75ca4d6e74516d0b5f3b9eb3ea45c76 Mon Sep 17 00:00:00 2001 From: Hongshun Wang Date: Wed, 9 Sep 2026 17:12:26 +0800 Subject: [PATCH 3/3] hongshun's advice --- .../apache/fluss/client/admin/FlussAdmin.java | 8 +- .../admin/CustomFlussClusterITCase.java | 129 ---------- .../fluss/client/admin/FlussAdminITCase.java | 75 ++++-- .../testutils/FlussClusterExtension.java | 19 +- .../testutils/TestingServerRestartUtils.java | 229 ++++++++++++++++++ tools/maven/suppressions.xml | 1 + 6 files changed, 304 insertions(+), 157 deletions(-) create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/testutils/TestingServerRestartUtils.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 5ed0a8af7c9..08738ce955a 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 @@ -33,6 +33,7 @@ import org.apache.fluss.config.cluster.AlterConfig; import org.apache.fluss.config.cluster.ConfigEntry; import org.apache.fluss.exception.FlussRuntimeException; +import org.apache.fluss.exception.InvalidServerTypeException; import org.apache.fluss.exception.LeaderNotAvailableException; import org.apache.fluss.exception.NotCoordinatorLeaderException; import org.apache.fluss.exception.RetriableException; @@ -164,7 +165,8 @@ public FlussAdmin(RpcClient client, MetadataUpdater metadataUpdater) { metadataUpdater::getCoordinatorServer, client, AdminGateway.class); // Refresh metadata for recoverable failures, but don't retry generic network errors because // a non-idempotent write may already have succeeded. NotCoordinatorLeaderException is safe - // to retry because the standby rejects the request before invoking the coordinator API. + // to retry because the standby rejects the request before invoking the coordinator API, + // and InvalidServerTypeException is raised during the handshake before sending the request. this.gateway = RetryableGatewayClientProxy.createRetryableGatewayProxy( rawGateway, @@ -173,7 +175,9 @@ public FlussAdmin(RpcClient client, MetadataUpdater metadataUpdater) { cause -> cause instanceof NotCoordinatorLeaderException || cause instanceof RetriableException, - NotCoordinatorLeaderException.class::isInstance, + cause -> + cause instanceof NotCoordinatorLeaderException + || cause instanceof InvalidServerTypeException, AdminGateway.class); AdminGateway rawReadOnlyGateway = GatewayClientProxy.createGatewayProxy( diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/CustomFlussClusterITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/CustomFlussClusterITCase.java index 605be8e79ab..c5595c5d43d 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/CustomFlussClusterITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/CustomFlussClusterITCase.java @@ -26,8 +26,6 @@ import org.apache.fluss.client.table.scanner.log.LogScanner; import org.apache.fluss.client.table.scanner.log.ScanRecords; import org.apache.fluss.client.table.writer.UpsertWriter; -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.config.MemorySize; @@ -39,26 +37,18 @@ import org.apache.fluss.metadata.TablePath; import org.apache.fluss.record.ChangeType; import org.apache.fluss.row.InternalRow; -import org.apache.fluss.server.coordinator.CoordinatorServer; import org.apache.fluss.server.testutils.FlussClusterExtension; import org.apache.fluss.server.zk.ZooKeeperClient; -import org.apache.fluss.server.zk.data.CoordinatorAddress; -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.types.RowType; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.io.TempDir; -import java.nio.file.Path; import java.time.Duration; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -70,107 +60,11 @@ import static org.apache.fluss.testutils.DataTestUtils.row; import static org.apache.fluss.testutils.InternalRowAssert.assertThatRow; import static org.apache.fluss.testutils.common.CommonTestUtils.retry; -import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil; import static org.assertj.core.api.Assertions.assertThat; /** IT case for tests that require manual cluster management. */ class CustomFlussClusterITCase { - @Test - void testAdminWriteRecoversAfterCoordinatorFailover(@TempDir Path tempDir) throws Exception { - final FlussClusterExtension flussClusterExtension = - FlussClusterExtension.builder().setNumOfTabletServers(1).build(); - CoordinatorServer standbyCoordinator = null; - try { - flussClusterExtension.start(); - CoordinatorServer firstLeader = flussClusterExtension.getCoordinatorServer(); - String zooKeeperConnectString = - firstLeader - .getZooKeeperClient() - .getCuratorClient() - .getZookeeperClient() - .getCurrentConnectionString(); - - Configuration standbyConf = new Configuration(); - standbyConf.setString(ConfigOptions.ZOOKEEPER_ADDRESS, zooKeeperConnectString); - standbyConf.setString(ConfigOptions.BIND_LISTENERS, "FLUSS://localhost:0"); - standbyConf.set( - ConfigOptions.REMOTE_DATA_DIR, - tempDir.resolve("standby-remote-data").toString()); - standbyCoordinator = new CoordinatorServer(standbyConf); - standbyCoordinator.start(); - - waitUntil( - () -> - flussClusterExtension - .getZooKeeperClient() - .getCoordinatorServerList() - .size() - == 2, - Duration.ofSeconds(30), - "Standby coordinator did not register"); - - try (Connection connection = - ConnectionFactory.createConnection( - flussClusterExtension.getClientConfig()); - Admin admin = connection.getAdmin()) { - String databaseName = "test_admin_write_after_coordinator_failover"; - admin.createDatabase(databaseName, DatabaseDescriptor.EMPTY, false).get(); - assertThat(admin.listDatabases().get()).contains(databaseName); - - killZooKeeperSession(firstLeader, zooKeeperConnectString); - CoordinatorServer newLeader = standbyCoordinator; - waitUntil( - () -> { - CoordinatorAddress leaderAddress = - flussClusterExtension - .getZooKeeperClient() - .getCoordinatorLeaderAddress() - .orElse(null); - return leaderAddress != null - && leaderAddress.getId().equals(newLeader.getServerId()) - && newLeader.getCoordinatorService().isLeader(); - }, - Duration.ofMinutes(1), - "Standby coordinator did not become leader"); - - Endpoint newLeaderEndpoint = - newLeader.getRpcServer().getBindEndpoints().stream() - .filter( - endpoint -> - endpoint.getListenerName() - .equals( - ConfigOptions.INTERNAL_LISTENER_NAME - .defaultValue())) - .findFirst() - .orElseThrow(IllegalStateException::new); - waitUntil( - () -> { - ServerNode cachedCoordinator = - flussClusterExtension - .getTabletServerById(0) - .getMetadataCache() - .getCoordinatorServer( - ConfigOptions.INTERNAL_LISTENER_NAME - .defaultValue()); - return cachedCoordinator != null - && cachedCoordinator.host().equals(newLeaderEndpoint.getHost()) - && cachedCoordinator.port() == newLeaderEndpoint.getPort(); - }, - Duration.ofSeconds(30), - "Tablet server did not learn the new coordinator leader"); - - admin.dropDatabase(databaseName, false, false).get(); - assertThat(admin.listDatabases().get()).doesNotContain(databaseName); - } - } finally { - if (standbyCoordinator != null) { - standbyCoordinator.close(); - } - flussClusterExtension.close(); - } - } - @Test void testProjectionPushdownWithEmptyBatches() throws Exception { Configuration conf = initConfig(); @@ -432,27 +326,4 @@ protected static Configuration initConfig() { conf.set(ConfigOptions.NETTY_CLIENT_NUM_NETWORK_THREADS, 1); return conf; } - - private static void killZooKeeperSession( - CoordinatorServer server, String zooKeeperConnectString) throws Exception { - CuratorFramework curatorClient = server.getZooKeeperClient().getCuratorClient(); - ZooKeeper zooKeeper = curatorClient.getZookeeperClient().getZooKeeper(); - CountDownLatch connectedLatch = new CountDownLatch(1); - ZooKeeper duplicateSession = - new ZooKeeper( - zooKeeperConnectString, - 1000, - event -> { - if (event.getState() == Watcher.Event.KeeperState.SyncConnected) { - connectedLatch.countDown(); - } - }, - zooKeeper.getSessionId(), - zooKeeper.getSessionPasswd()); - if (!connectedLatch.await(10, TimeUnit.SECONDS)) { - duplicateSession.close(); - throw new IllegalStateException("Failed to connect duplicate ZooKeeper session"); - } - duplicateSession.close(); - } } diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java index 70ad760f0e6..f7459f63306 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java @@ -85,6 +85,9 @@ import org.apache.fluss.server.replica.Replica; import org.apache.fluss.server.tablet.TestTabletServerGateway; import org.apache.fluss.server.testutils.FlussClusterExtension; +import org.apache.fluss.server.testutils.TestingServerRestartUtils; +import org.apache.fluss.server.testutils.TestingServerRestartUtils.RestartScenario; +import org.apache.fluss.server.testutils.TestingServerRestartUtils.RestartTarget; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.data.ServerTags; import org.apache.fluss.types.DataTypeChecks; @@ -92,6 +95,8 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; import javax.annotation.Nullable; @@ -1211,8 +1216,37 @@ void testListPartitionInfos() throws Exception { } } - @Test - void testListPartitionInfosAfterTabletServerRestart() throws Exception { + @ParameterizedTest + @EnumSource(RestartScenario.class) + void testCreateTableAfterCoordinatorServerRestart(RestartScenario restartScenario) + throws Exception { + TablePath tablePath = + TablePath.of( + DEFAULT_TABLE_PATH.getDatabaseName(), + "test_create_table_after_coordinator_server_restart"); + ZooKeeperClient zkClient = FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(); + + TestingServerRestartUtils.restartServers( + FLUSS_CLUSTER_EXTENSION, RestartTarget.COORDINATOR, restartScenario); + + if (restartScenario == RestartScenario.NEW_PORT) { + assertThatThrownBy( + () -> + admin.createTable(tablePath, DEFAULT_TABLE_DESCRIPTOR, false) + .get()) + .isInstanceOf(ExecutionException.class) + .hasCauseInstanceOf(NetworkException.class); + assertThat(zkClient.tableExist(tablePath)).isFalse(); + } + + admin.createTable(tablePath, DEFAULT_TABLE_DESCRIPTOR, true).get(); + assertThat(zkClient.tableExist(tablePath)).isTrue(); + } + + @ParameterizedTest + @EnumSource(RestartScenario.class) + void testListPartitionInfosAfterTabletServerRestart(RestartScenario restartScenario) + throws Exception { String dbName = DEFAULT_TABLE_PATH.getDatabaseName(); TablePath partitionedTablePath = TablePath.of(dbName, "test_retry_partitioned_table"); admin.createTable(partitionedTablePath, DATA1_PARTITIONED_TABLE_DESCRIPTOR, true).get(); @@ -1223,12 +1257,8 @@ void testListPartitionInfosAfterTabletServerRestart() throws Exception { admin.listPartitionInfos(partitionedTablePath).get(); assertThat(partitionInfosBefore).isNotEmpty(); - // Restart all tablet servers (they bind to new ports, making cached addresses stale). - for (int i = 0; i < FLUSS_CLUSTER_EXTENSION.getTabletServerNodes().size(); i++) { - FLUSS_CLUSTER_EXTENSION.stopTabletServer(i); - FLUSS_CLUSTER_EXTENSION.startTabletServer(i); - } - FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata(); + TestingServerRestartUtils.restartServers( + FLUSS_CLUSTER_EXTENSION, RestartTarget.TABLET_SERVERS, restartScenario); // Second query using the same admin client should succeed after retry with metadata // refresh (verifies RetryableGatewayClientProxy convergence on stale addresses). @@ -1237,22 +1267,28 @@ void testListPartitionInfosAfterTabletServerRestart() throws Exception { assertThat(partitionInfosAfter).hasSize(partitionInfosBefore.size()); } - @Test - void testKvSnapshotLeaseAfterCoordinatorServerRestart() throws Exception { + @ParameterizedTest + @EnumSource(RestartScenario.class) + void testKvSnapshotLeaseAfterCoordinatorServerRestart(RestartScenario restartScenario) + throws Exception { long tableId = admin.getTableInfo(DEFAULT_TABLE_PATH).get().getTableId(); TableBucket tableBucket = new TableBucket(tableId, 0); Map snapshots = Collections.singletonMap(tableBucket, 0L); - KvSnapshotLease lease = admin.createKvSnapshotLease("test-retry-kv-snapshot-lease", 60000L); + KvSnapshotLease lease = + admin.createKvSnapshotLease( + "test-retry-kv-snapshot-lease-" + restartScenario.name(), 60000L); ZooKeeperClient zkClient = FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(); // Restart the coordinator server so that the lease uses a stale cached address. - restartCoordinatorServer(zkClient); + TestingServerRestartUtils.restartServers( + FLUSS_CLUSTER_EXTENSION, RestartTarget.COORDINATOR, restartScenario); lease.acquireSnapshots(snapshots).get(); assertThat(zkClient.getKvSnapshotLeaseMetadata(lease.leaseId())).isPresent(); // Verify that release also refreshes metadata and retries against the new coordinator. - restartCoordinatorServer(zkClient); + TestingServerRestartUtils.restartServers( + FLUSS_CLUSTER_EXTENSION, RestartTarget.COORDINATOR, restartScenario); lease.releaseSnapshots(Collections.singleton(tableBucket)).get(); assertThat(zkClient.getKvSnapshotLeaseMetadata(lease.leaseId())).isNotPresent(); @@ -1261,22 +1297,13 @@ void testKvSnapshotLeaseAfterCoordinatorServerRestart() throws Exception { lease.acquireSnapshots(snapshots).get(); assertThat(zkClient.getKvSnapshotLeaseMetadata(lease.leaseId())).isPresent(); - restartCoordinatorServer(zkClient); - FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata(); + TestingServerRestartUtils.restartServers( + FLUSS_CLUSTER_EXTENSION, RestartTarget.COORDINATOR, restartScenario); lease.dropLease().get(); assertThat(zkClient.getKvSnapshotLeaseMetadata(lease.leaseId())).isNotPresent(); } - private void restartCoordinatorServer(ZooKeeperClient zkClient) throws Exception { - FLUSS_CLUSTER_EXTENSION.stopCoordinatorServer(); - waitUntil( - () -> !zkClient.getCoordinatorLeaderAddress().isPresent(), - Duration.ofMinutes(1), - "Coordinator server node still exists in ZooKeeper"); - FLUSS_CLUSTER_EXTENSION.startCoordinatorServer(); - } - @Test void testListPartitionInfosByPartitionSpec() throws Exception { String dbName = DEFAULT_TABLE_PATH.getDatabaseName(); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java index 0cc615482ab..d226817a096 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java @@ -263,11 +263,16 @@ public void close() throws Exception { /** Start a coordinator server. start a new one if no coordinator server exists. */ public void startCoordinatorServer() throws Exception { + startCoordinatorServer(coordinatorServerListeners); + } + + /** Start a coordinator server with the given listeners. */ + public void startCoordinatorServer(String bindListeners) throws Exception { if (coordinatorServer == null) { // if no coordinator server exists, create a new coordinator server and start Configuration conf = new Configuration(clusterConf); conf.setString(ConfigOptions.ZOOKEEPER_ADDRESS, zooKeeperServer.getConnectString()); - conf.setString(ConfigOptions.BIND_LISTENERS, coordinatorServerListeners); + conf.setString(ConfigOptions.BIND_LISTENERS, bindListeners); setRemoteDataDir(conf); setRemoteDataDirs(conf); coordinatorServer = new CoordinatorServer(conf, clock); @@ -310,6 +315,16 @@ public void startTabletServer(int serverId) throws Exception { startTabletServer(serverId, false); } + /** Start a new tablet server with the given listeners. */ + public void startTabletServer(int serverId, String bindListeners) throws Exception { + if (tabletServers.containsKey(serverId)) { + throw new IllegalArgumentException("Tablet server " + serverId + " already exists."); + } + Configuration overwriteConfig = new Configuration(); + overwriteConfig.setString(ConfigOptions.BIND_LISTENERS, bindListeners); + startTabletServer(serverId, overwriteConfig); + } + public void startTabletServer(int serverId, boolean forceStartIfExists) throws Exception { if (tabletServers.containsKey(serverId)) { if (!forceStartIfExists) { @@ -317,7 +332,7 @@ public void startTabletServer(int serverId, boolean forceStartIfExists) throws E "Tablet server " + serverId + " already exists."); } } - startTabletServer(serverId, null); + startTabletServer(serverId, (Configuration) null); } private void startTabletServer(int serverId, @Nullable Configuration overwriteConfig) diff --git a/fluss-server/src/test/java/org/apache/fluss/server/testutils/TestingServerRestartUtils.java b/fluss-server/src/test/java/org/apache/fluss/server/testutils/TestingServerRestartUtils.java new file mode 100644 index 00000000000..2b55bffd718 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/testutils/TestingServerRestartUtils.java @@ -0,0 +1,229 @@ +/* + * 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.testutils; + +import org.apache.fluss.cluster.ServerNode; +import org.apache.fluss.server.coordinator.CoordinatorServer; +import org.apache.fluss.server.zk.ZooKeeperClient; +import org.apache.fluss.utils.IOUtils; +import org.apache.fluss.utils.NetUtils; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; + +import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil; +import static org.assertj.core.api.Assertions.assertThat; + +/** Utilities for restarting testing servers with deterministic endpoint changes. */ +public final class TestingServerRestartUtils { + + /** Endpoint topology used when restarting testing servers. */ + public enum RestartScenario { + NEW_PORT, + SWAPPED_COORDINATOR_AND_TABLET_SERVER_PORTS + } + + /** Server group targeted by a {@link RestartScenario#NEW_PORT} restart. */ + public enum RestartTarget { + COORDINATOR, + TABLET_SERVERS + } + + private TestingServerRestartUtils() {} + + /** + * Restarts testing servers with deterministic endpoint changes and waits for ZooKeeper and + * server metadata to converge. + */ + public static void restartServers( + FlussClusterExtension extension, + RestartTarget restartTarget, + RestartScenario restartScenario) + throws Exception { + ZooKeeperClient zkClient = extension.getZooKeeperClient(); + switch (restartScenario) { + case NEW_PORT: + if (restartTarget == RestartTarget.COORDINATOR) { + restartCoordinatorServerWithNewPort(extension, zkClient); + } else { + restartTabletServersWithNewPorts(extension); + } + break; + case SWAPPED_COORDINATOR_AND_TABLET_SERVER_PORTS: + restartCoordinatorAndTabletServersWithSwappedPorts(extension, zkClient); + break; + default: + throw new IllegalArgumentException( + "Unsupported restart scenario: " + restartScenario); + } + } + + private static void restartCoordinatorServerWithNewPort( + FlussClusterExtension extension, ZooKeeperClient zkClient) throws Exception { + CoordinatorServer previousServer = extension.getCoordinatorServer(); + ServerNode previousNode = extension.getCoordinatorServerNode(); + try (NetUtils.Port newPort = NetUtils.getAvailablePort()) { + stopCoordinatorServer(extension, zkClient); + extension.startCoordinatorServer(bindListener(newPort.getPort())); + + ServerNode restartedNode = extension.getCoordinatorServerNode(); + assertThat(extension.getCoordinatorServer()).isNotSameAs(previousServer); + assertThat(restartedNode.uid()).isEqualTo(previousNode.uid()); + assertThat(restartedNode.host()).isEqualTo(previousNode.host()); + assertThat(restartedNode.port()) + .isEqualTo(newPort.getPort()) + .isNotEqualTo(previousNode.port()); + } + extension.waitUntilAllGatewayHasSameMetadata(); + } + + private static void restartTabletServersWithNewPorts(FlussClusterExtension extension) + throws Exception { + List previousNodes = extension.getTabletServerNodes(); + List newPorts = reservePorts(previousNodes.size()); + try { + for (int i = 0; i < previousNodes.size(); i++) { + ServerNode previousNode = previousNodes.get(i); + extension.stopTabletServer(previousNode.id()); + extension.startTabletServer( + previousNode.id(), bindListener(newPorts.get(i).getPort())); + } + extension.waitUntilAllGatewayHasSameMetadata(); + + for (int i = 0; i < previousNodes.size(); i++) { + ServerNode previousNode = previousNodes.get(i); + ServerNode restartedNode = getTabletServerNode(extension, previousNode.id()); + assertThat(restartedNode.uid()).isEqualTo(previousNode.uid()); + assertThat(restartedNode.host()).isEqualTo(previousNode.host()); + assertThat(restartedNode.port()) + .isEqualTo(newPorts.get(i).getPort()) + .isNotEqualTo(previousNode.port()); + } + } finally { + IOUtils.closeAllQuietly(newPorts); + } + } + + private static void restartCoordinatorAndTabletServersWithSwappedPorts( + FlussClusterExtension extension, ZooKeeperClient zkClient) throws Exception { + CoordinatorServer previousCoordinatorServer = extension.getCoordinatorServer(); + ServerNode previousCoordinator = extension.getCoordinatorServerNode(); + List previousTabletServers = extension.getTabletServerNodes(); + ServerNode swappedTabletServer = + previousTabletServers.stream() + .filter(tabletServer -> tabletServer.id() == 0) + .findFirst() + .orElseThrow( + () -> new IllegalStateException("Tablet server 0 does not exist.")); + List otherTabletServerPorts = reservePorts(previousTabletServers.size() - 1); + + try { + // Stop tablet servers while the coordinator is still available for controlled + // shutdown, then stop the coordinator before reusing their ports. + for (ServerNode tabletServer : previousTabletServers) { + extension.stopTabletServer(tabletServer.id()); + } + waitUntil( + () -> zkClient.getSortedTabletServerList().length == 0, + Duration.ofMinutes(1), + "Tablet server nodes still exist in ZooKeeper"); + stopCoordinatorServer(extension, zkClient); + + extension.startCoordinatorServer(bindListener(swappedTabletServer.port())); + extension.startTabletServer( + swappedTabletServer.id(), bindListener(previousCoordinator.port())); + int newPortIndex = 0; + for (ServerNode tabletServer : previousTabletServers) { + if (tabletServer.id() != swappedTabletServer.id()) { + extension.startTabletServer( + tabletServer.id(), + bindListener(otherTabletServerPorts.get(newPortIndex).getPort())); + newPortIndex++; + } + } + extension.waitUntilAllGatewayHasSameMetadata(); + + ServerNode restartedCoordinator = extension.getCoordinatorServerNode(); + ServerNode restartedTabletServer = + getTabletServerNode(extension, swappedTabletServer.id()); + assertThat(extension.getCoordinatorServer()).isNotSameAs(previousCoordinatorServer); + assertThat(restartedCoordinator.uid()).isEqualTo(previousCoordinator.uid()); + assertThat(restartedCoordinator.host()).isEqualTo(swappedTabletServer.host()); + assertThat(restartedCoordinator.port()) + .isEqualTo(swappedTabletServer.port()) + .isNotEqualTo(previousCoordinator.port()); + assertThat(restartedTabletServer.uid()).isEqualTo(swappedTabletServer.uid()); + assertThat(restartedTabletServer.host()).isEqualTo(previousCoordinator.host()); + assertThat(restartedTabletServer.port()) + .isEqualTo(previousCoordinator.port()) + .isNotEqualTo(swappedTabletServer.port()); + + newPortIndex = 0; + for (ServerNode tabletServer : previousTabletServers) { + if (tabletServer.id() != swappedTabletServer.id()) { + ServerNode restartedNode = getTabletServerNode(extension, tabletServer.id()); + assertThat(restartedNode.uid()).isEqualTo(tabletServer.uid()); + assertThat(restartedNode.host()).isEqualTo(tabletServer.host()); + assertThat(restartedNode.port()) + .isEqualTo(otherTabletServerPorts.get(newPortIndex).getPort()) + .isNotEqualTo(tabletServer.port()); + newPortIndex++; + } + } + } finally { + IOUtils.closeAllQuietly(otherTabletServerPorts); + } + } + + private static void stopCoordinatorServer( + FlussClusterExtension extension, ZooKeeperClient zkClient) throws Exception { + extension.stopCoordinatorServer(); + waitUntil( + () -> !zkClient.getCoordinatorLeaderAddress().isPresent(), + Duration.ofMinutes(1), + "Coordinator server node still exists in ZooKeeper"); + } + + private static List reservePorts(int portCount) { + List ports = new ArrayList<>(portCount); + try { + for (int i = 0; i < portCount; i++) { + ports.add(NetUtils.getAvailablePort()); + } + return ports; + } catch (RuntimeException e) { + IOUtils.closeAllQuietly(ports); + throw e; + } + } + + private static ServerNode getTabletServerNode(FlussClusterExtension extension, int serverId) { + return extension.getTabletServerNodes().stream() + .filter(serverNode -> serverNode.id() == serverId) + .findFirst() + .orElseThrow( + () -> + new IllegalArgumentException( + "Tablet server " + serverId + " does not exist.")); + } + + private static String bindListener(int port) { + return String.format("FLUSS://localhost:%d", port); + } +} diff --git a/tools/maven/suppressions.xml b/tools/maven/suppressions.xml index 694a6979b2b..1e65061f27f 100644 --- a/tools/maven/suppressions.xml +++ b/tools/maven/suppressions.xml @@ -23,6 +23,7 @@ +