[jira] [Comment Edited] (FLINK-31762) Subscribe to multiple Kafka topics may cause partition assignment skew

["Hongshun Wang (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-31762?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17908757#comment-17908757 ] 

Hongshun Wang edited comment on FLINK-31762 at 12/30/24 2:22 AM:
-----------------------------------------------------------------

Hi, [~lorenzo.affetti]  I'd like to do it. It's important for me and I also start a mail discussion, [https://lists.apache.org/list.html?dev@flink.apache.org]

 

Besides the problem in this Jira , I aslo met another problem: If I consume multiple topics from a source, the getSplitOwner will count the owner of  partitions separately for each topic. This means that there may be overlapping assignments. For example, if there are 3 Kafka topics with 16 partitions each, even if the Flink task's parallelism is 48, only 16 tasks may be assigned 3 partitions in the worst-case scenario.

 

for example, the inital is 3 parrelism for 3 partition.  (the startIndex = 3, thus (startIndex+ 1)%3 =0) 
||subtask||0||1 ||2||
||topic 1||1||2||3||

then change to 6 parrelism: 
||subtask||0||1||2 ||3||4||5||
||topic 1||1||2||3|| || || ||

then the new discovered partitions will be:  (startIndex +4)%6 = 7%6 = 1
||subtask||0||1 ||2||3||4||5||
||topic 1||1||2,4||3,5||6|| || ||

 


was (Author: JIRAUSER298968):
Hi, [~lorenzo.affetti]  I'd like to do it. It's important for me and I also start a mail discussion, [https://lists.apache.org/list.html?dev@flink.apache.org]

 

> 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: Lorenzo Affetti
>            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)