Skip to content

[client] Recover Admin writes after coordinator failover #4200

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sakshichitnis27 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

[client] Recover Admin writes after coordinator failover #4200

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a3c65cd
[client] Recover Admin writes after coordinator failover
sakshichitnis27 Sep 1, 2026
b66d60a
[client] Separate metadata refresh from write retry
sakshichitnis27 Sep 3, 2026
2844cec
hongshun's advice
Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 31 additions & 7 deletions fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 =

@loserwang1024 loserwang1024 Sep 2, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have two suggestions:

  1. I previously implemented this in [PR #3390]([client] Fix stale metadata on readOnlyGateway by adding RetryableGatewayClientProxy #3390), but a reviewer reminded me that write operations are not idempotent, so we should not retry them automatically. I’m thinking that we could still return an error without retrying, but refresh the metadata before doing so. This way, the operation can recover the next time the user retries it manually.

  2. With the approach described in point 1, we should not limit metadata refresh to cases where the RPC response contains a NotCoordinatorLeaderException. During an upgrade, the old CoordinatorServer’s IP address is not necessarily reused by a TabletServer. If there are spare IP addresses, the old IP may remain unused, in which case the request may fail with a NetworkException instead.

@litiliu , WDYT?

@litiliu litiliu Sep 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @loserwang1024

On point 2 (don't limit refresh to NotCoordinatorLeaderException): agreed. After a failover the old coordinator may be gone or its IP not reused, so the write can fail with NetworkException/TimeoutException instead. We should refresh cluster metadata (and drop the stale coordinator connection) on any failure so the client can recover.

On point 1 (writes are non-idempotent, don't auto-retry): agreed in general, but NotCoordinatorLeaderException is a special, safe case. In FlussRequestHandler#processRequest the leader check runs before the write method is invoked:

            if (isCoordinator && api.getApiKey() != ApiKeys.API_VERSIONS) {
                if (!((CoordinatorGateway) service).isLeader()) {
                    request.fail(
                            new NotCoordinatorLeaderException(
                                    "This coordinator server is not the current leader."));
                    return;
                }
            }

So this exception guarantees the mutation was rejected before execution — retrying it cannot duplicate a write. NetworkException/TimeoutException may already have executed (lost response), so those must NOT be auto-retried.

Proposed policy for the write gateway:

On any failure → refresh metadata + discard the stale coordinator connection (recovers the NetworkException/upgrade case; the user's next manual retry then succeeds).
Auto-retry once only for NotCoordinatorLeaderException (provably safe; better UX for the standby-alive case).
This keeps auto-retry strictly to the provably-safe error while still refreshing metadata for everything else. WDYT?

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);
Expand All @@ -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<List<ServerNode>> getServerNodes() {
CompletableFuture<List<ServerNode>> future = new CompletableFuture<>();
Expand Down
75 changes: 51 additions & 24 deletions fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,13 +85,18 @@
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;
import org.apache.fluss.types.DataTypes;

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;

Expand Down Expand Up @@ -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();
Expand All @@ -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).
Expand All @@ -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<TableBucket, Long> 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();
Expand All @@ -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();
Expand Down
64 changes: 57 additions & 7 deletions fluss-rpc/src/main/java/org/apache/fluss/rpc/RetryableGatewayClientProxy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -68,6 +69,8 @@ public class RetryableGatewayClientProxy implements InvocationHandler {
private final Object delegate;
private final Runnable metadataRefreshAction;
private final Executor refreshExecutor;
private final Predicate<Throwable> refreshPredicate;
private final Predicate<Throwable> retryPredicate;

/**
* Holds the currently in-flight metadata refresh, if any. Concurrent retriers piggyback on this
Expand All @@ -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<Throwable> refreshPredicate,
Predicate<Throwable> retryPredicate) {
this.delegate = delegate;
this.metadataRefreshAction = metadataRefreshAction;
this.refreshExecutor = refreshExecutor;
this.refreshPredicate = refreshPredicate;
this.retryPredicate = retryPredicate;
}

/**
Expand All @@ -102,6 +111,35 @@ public static <T extends RpcGateway> T createRetryableGatewayProxy(
Runnable metadataRefreshAction,
Executor refreshExecutor,
Class<T> 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 <T> the gateway type
* @return a retryable gateway proxy
*/
public static <T extends RpcGateway> T createRetryableGatewayProxy(
T delegate,
Runnable metadataRefreshAction,
Executor refreshExecutor,
Predicate<Throwable> refreshPredicate,
Predicate<Throwable> retryPredicate,
Class<T> gatewayClass) {
ClassLoader classLoader = gatewayClass.getClassLoader();

@SuppressWarnings("unchecked")
Expand All @@ -111,7 +149,11 @@ public static <T extends RpcGateway> T createRetryableGatewayProxy(
classLoader,
new Class<?>[] {gatewayClass},
new RetryableGatewayClientProxy(
delegate, metadataRefreshAction, refreshExecutor));
delegate,
metadataRefreshAction,
refreshExecutor,
refreshPredicate,
retryPredicate));
return proxy;
}

Expand Down Expand Up @@ -143,22 +185,30 @@ private <T> CompletableFuture<T> 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.<T>invokeWithRetry(
method, args, false))
shouldRetry
? RetryableGatewayClientProxy.this
.<T>invokeWithRetry(method, args, false)
: future)
.whenComplete(
(retryResult, retryError) -> {
if (retryError != null) {
Expand Down
62 changes: 62 additions & 0 deletions fluss-rpc/src/test/java/org/apache/fluss/rpc/RetryableGatewayClientProxyTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -148,6 +149,67 @@ public CompletableFuture<ApiVersionsResponse> 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<ApiVersionsResponse> 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<ApiVersionsResponse> apiVersions(
ApiVersionsRequest request) {
if (callCount.incrementAndGet() == 1) {
CompletableFuture<ApiVersionsResponse> 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);
Expand Down
Loading
Loading