[client] Admin write operations do not recover after coordinator leader failover

[litiliu]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

main (development)

Please describe the bug 🐞

Description

A long-lived Java Admin client keeps the coordinator node cached in MetadataUpdater. After coordinator leadership moves to a standby, Admin write operations such as dropDatabase are still sent to the old coordinator and fail with:

org.apache.fluss.exception.NotCoordinatorLeaderException:
This coordinator server is not the current leader.

The same client continues using the stale coordinator, so subsequent write attempts fail until the connection is recreated.

Reproduction

1. Start a Fluss cluster with two CoordinatorServers and at least one TabletServer.
2. Create one long-lived Java Connection / Admin client and initialize its metadata.
3. Create a non-default database.
4. Trigger coordinator failover so the cached coordinator becomes standby.
5. Using the same Admin instance, call dropDatabase (or another coordinator write operation).

Expected behavior

The client recognizes NotCoordinatorLeaderException, refreshes cluster metadata, resolves the new coordinator leader, and retries the request once.

Actual behavior

The future fails with NotCoordinatorLeaderException. The cached coordinator is not refreshed, and later Admin writes continue failing.

Root cause

FlussAdmin wraps only readOnlyGateway with RetryableGatewayClientProxy. The write gateway remains a raw GatewayClientProxy using metadataUpdater::getCoordinatorServer.

This was intentionally left out of #3389 / #3390 because generic network-error retries are unsafe for non-idempotent writes. However, NotCoordinatorLeaderException is different: FlussRequestHandler rejects the request before invoking the coordinator API, so retrying this specific failure cannot duplicate an already executed mutation.

Impact

All coordinator write APIs can remain unavailable to a long-lived client after coordinator failover, including database/table create, alter, and drop operations.

Solution

Add a narrow retry path for coordinator writes:

1. Retry only when the RPC response is NotCoordinatorLeaderException.
2. Refresh cluster metadata through MetadataUpdater.
3. Resolve the current coordinator leader and retry the same RPC once.
4. Do not retry generic NetworkException or TimeoutException for Admin writes, because execution may already have occurred.
5. Add a Coordinator HA integration test that keeps one Admin client open across leader failover and verifies a write operation succeeds afterward.

This could reuse the existing metadata-refresh/coalescing machinery while using a dedicated predicate for the write gateway rather than enabling all RetriableException retries.

Are you willing to submit a PR?

• I'm willing to submit a PR!