[coordinator] Withhold leader from UpdateMetadataRequest until NotifyLeaderAndIsr is acked

[damokelis]

Purpose

Linked issue: close #4149

The coordinator maintains two independently-updated views of replica state: TabletServerMetadataCache (pushed via UpdateMetadataRequest, determines what leader info clients see) and ReplicaManager.allReplicas (activated via NotifyLeaderAndIsrRequest, determines whether a tablet server can actually serve fetchLog/putKv for a bucket). These are sent as two separate RPCs (CoordinatorRequestBatch#sendRequestToTabletServers) with no atomicity guarantee between them.

Under normal conditions the window between "coordinator decided who the leader is" and "the target tablet server has admitted the replica" is milliseconds and rarely observable. For large tables, however, a restarting tablet server can take minutes to complete local log recovery before it processes NotifyLeaderAndIsrRequest, while UpdateMetadataRequest can be broadcast well before that (e.g. via CoordinatorEventProcessor#processNewTabletServer when the server re-registers). A client that picks up the new leader from metadata during this window sends requests to a server whose ReplicaManager still treats the bucket as NoneReplica, and is rejected with NotLeaderOrFollowerException/LeaderNotAvailableException in a tight retry loop until the server finally catches up (observed: single-digit minutes to over ten minutes for a large 12-bucket table). See #4149 for full details and reproduction steps.

Brief change log

• CoordinatorContext: expose isPendingLeaderActivation(TableBucket) on the existing (previously read-only, health-API-only) pendingLeaderActivationBuckets tracking.
• CoordinatorRequestBatch#addUpdateMetadataRequestForTabletServers: withhold a bucket's leader from UpdateMetadataRequest for as long as isPendingLeaderActivation is true. sendNotifyLeaderAndIsrRequest already runs strictly before sendUpdateMetadataRequest in the same batch, so the pending mark set for an activation round is always visible to the accompanying metadata push.
• CoordinatorEventProcessor#processNotifyLeaderAndIsrResponseReceivedEvent: when a pending bucket's activation is genuinely acknowledged, clear the mark and proactively push a follow-up UpdateMetadataRequest (rather than waiting for some unrelated coordinator event to trigger the next broadcast). Also clear the mark for buckets that go offline via onReplicaBecomeOffline, since a bucket whose NotifyLeaderAndIsrRequest comes back with a per-bucket error never reaches the ack path and would otherwise be stuck pending forever, permanently withholding its leader even after a successful re-election.
• CoordinatorRequestBatch#sendNotifyLeaderAndIsrRequest: on RPC send failure, deliberately leave the pending mark untouched (see "Notable subtlety" below).
• Also fixes an unrelated but adjacent plain Map.put overwrite in the same method (case3/4/10/11) that could silently discard already-queued non-empty bucket data for the same table within one UpdateMetadataRequest batch — now logged instead of silent.

Notable subtlety: an earlier iteration of this change also cleared the pending mark when the NotifyLeaderAndIsrRequest RPC itself failed to send (e.g. target server not yet reachable during a rolling restart), reasoning that this avoided a stale RED in the health API. That turned out to reintroduce the exact race being fixed here, just triggered by RPC failure instead of by slow log recovery: clearing the mark let the leader be advertised to clients while the target server's ReplicaManager had never actually admitted the replica. The current version keeps the mark untouched on send failure; a transient GetClusterHealth RED during this window is expected and reflects genuine uncertainty, not a bug.

Tests

• mvn -pl fluss-server -am test -Dtest='org.apache.fluss.server.coordinator.**': 352/353 passing. The sole unrelated failure (CoordinatorServerITCase#testRunServerUsingProcess) reproduces identically on an unmodified checkout and is a local-environment issue spawning a separate JVM process, not a regression from this change.
• CoordinatorRequestBatchTest#testNotifyLeaderAndIsrSendFailureLeavesLeaderPending (renamed/inverted from a prior version that asserted the opposite, now-incorrect, behavior): asserts the pending mark survives an RPC send failure.
• CoordinatorRequestBatchTest#testNotifyLeaderAndIsrSendFailureToFollowerDoesNotClearOtherPending: asserts a follower-send failure doesn't touch an unrelated bucket's pending mark.
• CoordinatorEventProcessorTest (existing testSchemaChange/testTableRegistrationChange etc.): exercise the proactive follow-up UpdateMetadataRequest push after a pending mark is cleared.
• Manually reproduced and verified against a real cluster: rolling-restarted three tablet servers hosting a large table (including one requiring ~11 minutes of local log recovery), with a client continuously reading from the affected buckets throughout. Before this fix, NotLeaderOrFollowerException/LeaderNotAvailableException recurred for minutes after all tablet servers reported ready. After this fix, the client's cumulative error count did not increase at all across the full rolling-restart window plus an additional 10-minute observation period.

API and Format

No public API or on-disk/wire format change. Behavior change is purely in when a bucket's leader becomes visible via UpdateMetadataRequest/metadata() — it is delayed until the leader is actually able to serve requests, never advertised earlier than before.

Documentation

No new user-facing feature; no documentation change needed.

Generative AI disclosure

• Yes (tool: Claude Code, Claude Sonnet model)

Root-cause investigation, code changes, and this PR/issue writeup were produced with Claude Code assistance. All changes were reviewed by a human, verified with the automated test suite, and independently validated against a live cluster (real rolling-restart reproduction described in the Tests section above) before submission.