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/2] [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/2] [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