[server] Push partition filtering to the server-side to reduce network bandwidth usage

[xiaochen-zhou]

Purpose

Push partition filtering to the server-side to minimize network bandwidth usage.

Brief change log

Tests

API and Format

Documentation

[wuchong]

Thanks @xiaochen-zhou for the contribution. I have some comments for the PR.

Please create an issue and reach consensus on the approach before submitting a pull request.

While this PR does enhance the code, I'm not sure it will significantly reduce network load in practice. The listPartitions call is made every 10 seconds, and typically there are fewer than 100 partitions in total — so the payload size is relatively small.

More importantly, the main bottleneck for this RPC lies in ZooKeeper performance, and since the partition filtering isn't pushed down to the ZooKeeper layer, the server still retrieves all partition data before applying the filter. As a result, we don’t get the expected reduction in backend load.

[wuchong]

We can convert the filter to partialPartitionSpec during the construction of FlinkSourceEnumerator to avoid the repeated conversion for listPartitions.

[xiaochen-zhou]

Thanks @xiaochen-zhou for the contribution. I have some comments for the PR.

Please create an issue and reach consensus on the approach before submitting a pull request.

While this PR does enhance the code, I'm not sure it will significantly reduce network load in practice. The listPartitions call is made every 10 seconds, and typically there are fewer than 100 partitions in total — so the payload size is relatively small.

More importantly, the main bottleneck for this RPC lies in ZooKeeper performance, and since the partition filtering isn't pushed down to the ZooKeeper layer, the server still retrieves all partition data before applying the filter. As a result, we don’t get the expected reduction in backend load.

Thanks for your review. updated the code. I will create an issue.

[xiaochen-zhou]

Thanks @xiaochen-zhou for the contribution. I have some comments for the PR.

Please create an issue and reach consensus on the approach before submitting a pull request.

While this PR does enhance the code, I'm not sure it will significantly reduce network load in practice. The listPartitions call is made every 10 seconds, and typically there are fewer than 100 partitions in total — so the payload size is relatively small.

More importantly, the main bottleneck for this RPC lies in ZooKeeper performance, and since the partition filtering isn't pushed down to the ZooKeeper layer, the server still retrieves all partition data before applying the filter. As a result, we don’t get the expected reduction in backend load.

I think we can optimize on the ZK side: when the ResolvedPartitionSpec is a complete partition spec, we can directly retrieve the partition instead of scanning all partitions. This can satisfy some scenarios.