[
https://issues.apache.org/jira/browse/FLINK-31762?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18119727#comment-18119727
]
Seungki Kim commented on FLINK-31762:
-------------------------------------
Hi [~loserwang1024], are you still working on this? If not, I'd be happy to
pick it up.
We hit the single-topic variant you described in your comment in production
(Flink 2.2.1, flink-connector-kafka 4.0.1-2.0; the assignment logic is
unchanged on main). We first rescaled a KafkaSource from 3 to 6, then grew the
topic from 3 to 6 partitions. The three restored splits kept the owners chosen
by operator-state redistribution, while the three new partitions were placed by
{code:java}
{{(topicHash + partition) % 6}} {code}
without regard to the current load. The readers ended up with 2/2/1/0/0/1
partitions: two subtasks stayed idle. A savepoint restart at the same
parallelism restored exactly the same skew. The only fix was rescaling 6 -> 3
-> 6.
Now that FLIP-537 (FLINK-38564) is in Flink 2.2 and DynamicKafkaSource already
uses it (FLINK-39012 / FLINK-39980), the building blocks seem to be there.
My proposal for the plain KafkaSource:
# Implement \{{SupportsSplitReassignmentOnRecovery}}. On recovery, the
enumerator collects the splits reported via
\{{ReaderInfo#getReportedSplitsOnRegistration()}} and redistributes them evenly
across the current readers.
# Assign newly discovered partitions to the least-loaded reader, instead of
\{{(topicHash + partition) % numReaders}}.
This also covers the multi-topic skew in the original description. The existing
\{{GlobalSplitOwnerAssigner}} logic could be reused.
# Since this changes assignment behaviour, I'd put it behind an option, with
the current hash-based assignment as the default. Happy to discuss the default.
If this direction sounds reasonable, I'll open a PR against
flink-connector-kafka. cc. [~arvid]
> Subscribe to multiple Kafka topics may cause partition assignment skew
> ----------------------------------------------------------------------
>
> Key: FLINK-31762
> URL: https://issues.apache.org/jira/browse/FLINK-31762
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / Kafka
> Affects Versions: 1.13.0, 1.18.0
> Reporter: Liam
> Assignee: Hongshun Wang
> Priority: Major
> Attachments: image-2023-04-11-08-00-16-054.png,
> image-2023-04-11-08-12-24-115.png
>
>
> To simplify the demonstration, let us assume that there are two topics, and
> each topic has four partitions. We have set the parallelism to eight to
> consume these two topics. However, the current partition assignment method
> may lead to some subtasks being assigned two partitions while others are left
> with none.
> !image-2023-04-11-08-00-16-054.png|width=500,height=143!
> In my case, the situation is even worse as I have ten topics, each with 100
> partitions. If I set the parallelism to 1000, some slots may be assigned
> seven partitions while others remain unassigned.
> To address this issue, I propose a new partition assignment solution. In this
> approach, round-robin assignment takes place between all topics, not just one.
> For example, the ideal assignment for the case mentioned above is presented
> below:
>
> !https://imgr.whimsical.com/object/A4jSJwgQNrc5mgpGddhghq|width=513,height=134!
> This new solution can also handle cases where each topic has more partitions.
> !image-2023-04-11-08-12-24-115.png|width=444,height=127!
> Let us work together to reach a consensus on this proposal. Thank you!
>
> FYI: how the partition be assigned currently
> {code:java}
> public class KafkaTopicPartitionAssigner {
> public static int assign(KafkaTopicPartition partition, int
> numParallelSubtasks) {
> return assign(partition.getTopic(), partition.getPartition(),
> numParallelSubtasks);
> } public static int assign(String topic, int partition, int
> numParallelSubtasks) {
> int startIndex = ((topic.hashCode() * 31) & 0x7FFFFFFF) %
> numParallelSubtasks; // here, the assumption is that the id of Kafka
> partitions are always ascending
> // starting from 0, and therefore can be used directly as the offset
> clockwise from the
> // start index
> return (startIndex + partition) % numParallelSubtasks;
> }
> {code}
> for Kafka Source, it's implemented in the KafkaSourceEnumerator as below
> {code:java}
> static int getSplitOwner(TopicPartition tp, int numReaders) {
> int startIndex = ((tp.topic().hashCode() * 31) & 0x7FFFFFFF) %
> numReaders; // here, the assumption is that the id of Kafka partitions
> are always ascending
> // starting from 0, and therefore can be used directly as the offset
> clockwise from the
> // start index
> return (startIndex + tp.partition()) % numReaders;
> } {code}
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
