apache / apache/fluss

[server][client] Partition expiration storms ZooKeeper via repeated GET_METADATA on non-existent partitions until enumerator unsubscribes

Open
#3,466 2 comments 0 reactions 1 assignee Claimed by @loserwang1024 View on GitHub
Dominant language
Java
Stars
2.1k
Forks
625
Avg merge
3d 14h
Merged PRs (30d)
97

Description

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.

### Fluss version

0.9.0 (latest release)

### Please describe the bug 🐞

### Problem
When a partition is dropped (e.g. by auto-partition expiration), in-flight Fluss readers/writers and the server-side metadata layer collectively generate a burst of ZooKeeper traffic that can last up to scan.partition.discovery.interval (default 1 minute). With moderate Flink parallelism this is enough to overwhelm the ZooKeeper ensemble.

### Reproduction shape
* Partitioned table with N partitions (e.g. 100), Flink source/sink parallelism P (e.g. 100).
* Some partitions are expired/dropped on the server side.
* Per-task MetadataUpdater (reader/writer) hits a PartitionNotExistException on its next operation against an expired partition, then immediately re-issues GET_METADATA to refresh.
* Each such GET_METADATA on the tablet server cannot find the partition in TabletServerMetadataCache and falls back to ZooKeeper via ZkBasedMetadataProvider#getPartitionsMetadataFromZK
The enumerator only learns about the deletion on the next discovery tick (scan.partition.discovery.interval, default 1 min). Only after it broadcasts PartitionsRemovedEvent will the readers unsubscribe and stop retrying.

### Amplification math
For P parallel subtasks × N missing partitions, every GET_METADATA round-trip to a tablet server fans out into ZK reads (one per missing partition path) because the path used by the cache is partitionName, not partitionId. With P=100 and N=100, this is on the order of ≥10,000 concurrent ZK gets, sustained for up to the discovery interval. This is consistent with the existing warning in [scan.partition.discovery.interval docs](file:///Users/loserwang/workstation/github/fluss/website/docs/engine-flink/options.md#L102) that the default cannot be lowered because of ZK pressure.

### Solution

### Root cause
* No negative cache for missing partitions. TabletServerMetadataCache#getPartitionMetadata returns Optional.empty() for an unknown partition, and ZkBasedMetadataProvider#getPartitionsMetadataFromZK is invoked again on every request, with no record that the partition was already proven absent.
* Per-request ZK fan-out. ZkBasedMetadataProvider#getPartitionsMetadataFromZK does getPartitionIds(partitionPaths) → one ZK get per partition path, then getBucketMetadataForPartitions(...) over the returned ids. If a single requested partition is missing, the whole batch throws PartitionNotExistException, but the ZK gets are still issued. There is no short-circuit using the in-cache partition set.
* Client side has no termination condition. Reader/writer MetadataUpdater retries indefinitely on PartitionNotExistException and Sender#handlePartitionNotExist and waits for the enumerator’s PartitionsRemovedEvent instead of self-evicting.

### Proposed fix

#### Server side (`fluss-server`)

- **Skip ZK lookup for partitions already known via cache.** In `getPartitionsMetadataFromZK`, partition the input set into `inCache` vs `missingFromCache` first; only the `missingFromCache` slice needs a ZK round-trip.
- **Introduce a bounded negative cache for confirmed non-existent partitions** (`PhysicalTablePath → tombstone`) in `TabletServerMetadataCache`.
- Subsequent `GET_METADATA` requests for the same path return `PartitionNotExistException` directly without touching ZK.
- **Critical correctness concern:** the tombstone must be invalidated when a partition is later (re)created. Options:
1. Drive invalidation from the existing `MetadataChangeEvent` / `ZkBasedMetadataNotifier` path so any partition-create notification clears the tombstone.
2. Bound TTL of the tombstone (short, e.g. a few seconds) so a re-created partition is at most briefly invisible, accepting an extra ZK get after expiry.
- Option (1) is preferred to fully avoid the "partition recreated but cached as missing" pitfall called out in the request.

#### Client side (`fluss-client`, `fluss-flink`)

- **On `PartitionNotExistException`, stop self-retrying that partition** in `MetadataUpdater` / `Sender` / log fetcher paths.
- Drop the in-flight batches for that partition immediately (writer) and stop subscribing buckets of that partition (reader).
- Surface the eviction proactively to the enumerator (e.g. via a new source event from reader → enumerator, or by piggy-backing on the next heartbeat) instead of waiting up to `scan.partition.discovery.interval`.
- This mirrors the negative cache on the server: the client should not keep asking for a partition that has been authoritatively reported as gone.

### Are you willing to submit a PR?

- [x] I'm willing to submit a PR!

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.