danny0405 commented on issue #20306: URL: https://github.com/apache/hudi/issues/20306#issuecomment-6159825739
The diagnosis matches the current implementation. [`RemotePartitionerHandler`](https://github.com/apache/hudi/blob/b86a34b59e015b68796d88de8b18eea38c1739ae/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/RemotePartitionerHandler.java) advances `nextIndex` by `numBuckets` modulo writer parallelism, and [`RemotePartitionHelper`](https://github.com/apache/hudi/blob/b86a34b59e015b68796d88de8b18eea38c1739ae/hudi-common/src/main/java/org/apache/hudi/common/util/RemotePartitionHelper.java) adds the bucket ID. For fixed `b=c=10`, this indeed sends the same bucket ID from every data partition to the same writer. The proposed rotation looks sound for fixed `b` and `c`. I checked the formula independently for all 4,096 `(b,c)` pairs with both values in `1..64`: it preserves each partition's per-writer bucket counts, gives every fixed bucket ID full writer coverage over `c` assignments, and is periodic with period `c` (hence the claim also holds for any consecutive window of that length). This is a formula check, not a Flink benchmark reproduction. A few implementation points worth covering: - **The response/cache needs enough information for the rotation.** Today the helper receives and caches only a start offset; the server's modulo-reduced `nextIndex` loses the assignment ordinal. In general, changing that offset alone cannot implement `(start + ((k + shift) % b)) % c`. For example, with `b=4,c=6,p=3`, the proposed mapping for buckets `0..3` is `[1,2,3,0]`, whereas simply adding `shift=1` to the offset gives `[1,2,3,4]` and changes the allocation positions. The server could cache an allocation containing both the start and shift, with the helper applying the bucket-local wrap. Both values should be assigned consistently once per new partition and reused across clients/retries. - **Keep the guarantee scoped to fixed bucket count and parallelism.** The Flink remote partitioner also accepts partition-specific bucket counts via `NumBucketsFunction`; for mixed counts, `start=p*b` no longer describes the existing allocator. That path needs an explicit policy and regression coverage. - **Test the real handler/helper path.** Besides `10/10`, cover a shared-factor case such as `4/6`, a coprime case, `b>c`, repeated requests, and concurrent first requests. Assert stable routing for each `(partitionPath,bucketId)` and preserved bucket counts as well as hot-bucket coverage; the current Flink partitioner tests mock the helper. This should address the reported cross-partition concentration when comparable hot-bucket traffic repeats across partitions. It does not guarantee equal record load for unequal partition volumes or split a hot bucket within one partition. A before/after run of the reported workload would be useful validation alongside the mapping tests. -- 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]
