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..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,7 +33,10 @@ 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; import org.apache.fluss.metadata.DatabaseChange; import org.apache.fluss.metadata.DatabaseDescriptor; import org.apache.fluss.metadata.DatabaseInfo; @@ -157,15 +160,25 @@ 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); + // 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, + // and InvalidServerTypeException is raised during the handshake before sending the request. + this.gateway = + RetryableGatewayClientProxy.createRetryableGatewayProxy( + rawGateway, + () -> refreshCoordinatorMetadata(client, metadataUpdater), + refreshExecutor, + cause -> + cause instanceof NotCoordinatorLeaderException + || cause instanceof RetriableException, + cause -> + cause instanceof NotCoordinatorLeaderException + || cause instanceof InvalidServerTypeException, + AdminGateway.class); AdminGateway rawReadOnlyGateway = GatewayClientProxy.createGatewayProxy( metadataUpdater::getRandomTabletServer, client, AdminGateway.class); @@ -178,6 +191,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/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-rpc/src/main/java/org/apache/fluss/rpc/RetryableGatewayClientProxy.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/RetryableGatewayClientProxy.java index cc368e05e12..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 @@ -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,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,10 +81,16 @@ 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; } /** @@ -102,6 +111,35 @@ public static T createRetryableGatewayProxy( Runnable metadataRefreshAction, Executor refreshExecutor, Class gatewayClass) { + return createRetryableGatewayProxy( + delegate, + metadataRefreshAction, + refreshExecutor, + RetriableException.class::isInstance, + RetriableException.class::isInstance, + gatewayClass); + } + + /** + * 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 + * @return a retryable gateway proxy + */ + public static T createRetryableGatewayProxy( + T delegate, + Runnable metadataRefreshAction, + Executor refreshExecutor, + Predicate refreshPredicate, + Predicate retryPredicate, + Class gatewayClass) { ClassLoader classLoader = gatewayClass.getClassLoader(); @SuppressWarnings("unchecked") @@ -111,7 +149,11 @@ public static T createRetryableGatewayProxy( classLoader, new Class[] {gatewayClass}, new RetryableGatewayClientProxy( - delegate, metadataRefreshAction, refreshExecutor)); + delegate, + metadataRefreshAction, + refreshExecutor, + refreshPredicate, + retryPredicate)); return proxy; } @@ -143,22 +185,30 @@ private CompletableFuture invokeWithRetry(Method method, Object[] args, b return; } Throwable cause = ExceptionUtils.stripCompletionException(throwable); - if (!(cause instanceof RetriableException) || !retry) { + 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 d4c8f9dc382..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 @@ -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,67 @@ public CompletableFuture apiVersions( assertThat(refreshCount.get()).isEqualTo(0); } + @Test + void testCustomPredicatesRefreshWithoutRetryingNetworkError() throws Exception { + AtomicInteger callCount = new AtomicInteger(0); + AtomicInteger refreshCount = new AtomicInteger(0); + + RpcGateway delegate = createGateway(callCount, 1); + RpcGateway proxy = + RetryableGatewayClientProxy.createRetryableGatewayProxy( + delegate, + refreshCount::incrementAndGet, + REFRESH_EXECUTOR, + NetworkException.class::isInstance, + 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(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 void testMetadataRefreshFailureDoesNotPreventRetry() throws Exception { AtomicInteger callCount = new AtomicInteger(0); 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 @@ +