Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/4301
@StephanEwen thanks for the review. Your suggestion makes a lot of sense.
I've fixed this up as the following:
- Have a new method `KafkaTopicAssigner.assign(KafkaTopicPartition
partition, int numSubtasks)` that defines a strict contract, such that when
locally used to filter out partitions, the resulting distribution of the
partitions of a single topic are guaranteed to be:
1. Uniformly distributed across subtasks
2. Partitions are round-robin distributed (strictly CLOCKWISE w.r.t.
ascending subtask indices) by using the partition id as the offset from a
starting index determined using the topic name. The extra directional contract
makes this more stricter than what we had before, which we may be round-robin
assigning partitions counter-clockwise. This should make the actual assignment
scheme much more predictable as perceived by the user, since they just need to
know the start index of a specific topic to understand how the partitions of
the topic are distributed across subtasks. We could add some log that states
the start index of the topics it is consuming.
- Strengthen the tests in `KafkaConsumerPartitionAssignmentTest` to test
this contract. Uniform distribution was already tested in that suite of tests,
the change makes the tests also verify the "clockwise round-robin since some
start index" contract.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
