[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?

[loserwang1024]

@sakshichitnis27 @litiliu I have adjust test, please help a final check.

[sakshichitnis27]

Thanks @loserwang1024. I reviewed commit 2844cec and the changes look good to me.
The additional InvalidServerTypeException retry is safe because the exception is raised during the API versions handshake before the Admin write request is sent. The restart scenarios also cover both a changed coordinator endpoint and coordinator/tablet-server port reuse.
I ran the affected proxy and Admin integration tests, Spotless, and the Java 8 compatibility build successfully. Thanks for improving the tests.
@litiliu, could you please help with the final check?

[litiliu]

@loserwang1024
Thanks for the updates. After comparing this PR with #4216 and discussing the connection handling in #4263, I’m planning to close #4216 and let this PR own the Admin write recovery.
I suggest separating the responsibilities as follows:

• [rpc] Handle endpoint changes for cached server connections #4263 handles endpoint changes at the RPC transport layer by identifying connections with UID + host + port.
• This PR handles metadata refresh and the safe retry policy for Admin writes.
  Once [rpc] Handle endpoint changes for cached server connections #4263 is merged, the explicit client.disconnect(coordinator.uid()) here should no longer be necessary. Refreshed metadata will naturally select the connection for the new endpoint. Moreover, disconnect(uid) in [rpc] Handle endpoint changes for cached server connections #4263 closes all connections belonging to that UID, which could also close a newly established connection.
  Therefore, the refresh action can be simplified to:
  metadataUpdater::refreshClusterUntilAvailable
  I think the current retry policy should otherwise be preserved:
• NotCoordinatorLeaderException: refresh and retry, because the write is rejected before execution.
• InvalidServerTypeException: refresh and retry, because it is raised during the handshake before the write request is sent.
• Other RetriableExceptions such as network errors or timeouts: refresh only, without automatic retry, because the write may already have been executed.
  I’d also suggest moving the integration test from [client] Recover admin writes after coordinator leader failover #4216 that performs an actual leader failover while the old Coordinator remains alive. It directly verifies the NotCoordinatorLeaderException recovery path, complementing this PR’s new-port and swapped-port scenarios.
  The intended merge order would be: merge [rpc] Handle endpoint changes for cached server connections #4263 first, rebase this PR, remove the explicit disconnect, and then merge this PR.