loserwang1024 opened a new issue, #3466: URL: https://github.com/apache/fluss/issues/3466
### 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! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
