[client] KV batch scans keep using a stale leader after failover #4909

Fluss version

main (1.1-SNAPSHOT), reproduced on b85258c.

Please describe the bug 🐞

After a bucket leader goes offline, recreating a KV batch scanner with the same Connection and Table can keep targeting the old leader, even after a new leader is ready.

Reproduction:

1. Create a primary-key table with one bucket and three replicas on a three-TabletServer cluster.
2. Write eight rows and scan them to cache leader A.
3. Stop A and wait until another replica becomes the ready leader.
4. Keep the original Connection and Table, and create a batch scanner. The scan fails with NetworkException.
5. Recreate the scanner once. It still targets A and fails.

KvBatchScanner does not invalidate stale routing on relevant failures. checkAndUpdateMetadata() only refreshes missing table or partition metadata, and leaderFor() returns a cached non-null leader without refreshing it.

Expected behavior: the failing scan should invalidate stale routing so that a caller-created scanner can resolve the new leader without replacing the Connection.

Solution

Invalidate the affected table or partition bucket metadata on metadata-related and lost-session failures, covering both open and continuation requests.

Keep the current scan terminal. Do not automatically reopen an established session, since doing so would replace its snapshot.

Related: #1885 fixes bucket-level metadata handling in LogFetcher. #4263 addresses cached RPC endpoint changes rather than KV scanner routing invalidation.

Are you willing to submit a PR?

• I'm willing to submit a PR!