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 2 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
2 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
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
34 changes: 27 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 @@ -34,6 +34,8 @@
import org.apache.fluss.config.cluster.ConfigEntry;
import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.exception.LeaderNotAvailableException;
import org.apache.fluss.exception.NotCoordinatorLeaderException;
import org.apache.fluss.exception.RetriableException;
import org.apache.fluss.metadata.DatabaseChange;
import org.apache.fluss.metadata.DatabaseDescriptor;
import org.apache.fluss.metadata.DatabaseInfo;
Expand Down Expand Up @@ -157,15 +159,22 @@ 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.
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,
NotCoordinatorLeaderException.class::isInstance,
AdminGateway.class);
AdminGateway rawReadOnlyGateway =
GatewayClientProxy.createGatewayProxy(
metadataUpdater::getRandomTabletServer, client, AdminGateway.class);
Expand All @@ -178,6 +187,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
129 changes: 129 additions & 0 deletions fluss-client/src/test/java/org/apache/fluss/client/admin/CustomFlussClusterITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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();
Expand Down Expand Up @@ -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();
}
}
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
Loading