[client] Recover Admin writes after coordinator failover

[sakshichitnis27]

Purpose

Linked issue: close #4027

A long-lived Java Admin client keeps using its cached coordinator connection after leadership moves to a standby. Coordinator write operations then continue to fail with NotCoordinatorLeaderException until the connection is recreated.

Brief change log

• Allow RetryableGatewayClientProxy callers to provide a retry predicate while preserving RetriableException as the default.
• Wrap the Admin write gateway with a retry policy limited to NotCoordinatorLeaderException.
• Refresh cluster metadata and discard the stale shared coordinator connection before retrying the write once.
• Add unit coverage proving the write policy excludes generic network failures.
• Add an HA integration test that keeps one Connection and Admin open across coordinator failover and verifies dropDatabase succeeds.

Tests

./mvnw -pl fluss-client -am -Dtest=RetryableGatewayClientProxyTest,CustomFlussClusterITCase#testAdminWriteRecoversAfterCoordinatorFailover -Dsurefire.failIfNoSpecifiedTests=false test

• RetryableGatewayClientProxyTest: 7 tests passed.
• Coordinator failover regression: 1 test passed.
• Checkstyle and Spotless checks passed as part of the Maven run.

API and Format

No public API or storage format changes. The new retry-predicate overload is internal.

Documentation

No documentation changes. This fixes existing Admin failover behavior.

[loserwang1024]

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]

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?

[litiliu]

Thanks @sakshichitnis27 for driving this, and @loserwang1024 for the review. Based on that feedback I put up an alternative implementation in #4216 for #4027, in case it's useful to the discussion.

Key differences that address the two review points:

• Split "refresh" from "retry". RetryableGatewayClientProxy takes separate refreshPredicate and retryPredicate. The Admin write gateway refreshes metadata on any recoverable error (NotCoordinatorLeaderException || RetriableException) so the NetworkException/upgrade case (old coordinator IP not reused) also recovers, but auto-retries only NotCoordinatorLeaderException — which FlussRequestHandler rejects before invoking the write API, so a retry can't duplicate a non-idempotent mutation. Other errors refresh metadata and surface the original error for a manual retry.
• Connection repointing. Instead of an explicit disconnect() in FlussAdmin, I made NettyClient recreate the connection when the address behind a uid changes (coordinators share cs-0). This also covers tablet-server rolling upgrades generically.

Happy to fold whichever direction the community prefers into a single PR — didn't mean to fragment the effort.

[sakshichitnis27]

Thanks @loserwang1024 and @litiliu. I’ve updated the PR to separate metadata refresh from automatic retry:

• Admin writes refresh coordinator metadata for NotCoordinatorLeaderException and recoverable failures such as NetworkException/TimeoutException.
• Only NotCoordinatorLeaderException is retried automatically, since it is rejected before the write operation executes.
• Network/timeout failures are returned to the caller after metadata refresh, allowing the next manual retry to use the current coordinator.
• The stale coordinator connection is disconnected after refresh.
  I kept the connection handling localized to FlussAdmin to keep this fix narrowly scoped to [client] Admin write operations do not recover after coordinator leader failover #4027.
  I also added unit coverage for both policies and reran the coordinator-failover integration test successfully. Could you please take another look?