[jira] [Created] (FLINK-29437) The partition of data before and after the Kafka Shuffle are not aligned

["Zakelly Lan (Jira)" <ji...@apache.org>]

Zakelly Lan created FLINK-29437:
-----------------------------------

             Summary: The partition of data before and after the Kafka Shuffle are not aligned
                 Key: FLINK-29437
                 URL: https://issues.apache.org/jira/browse/FLINK-29437
             Project: Flink
          Issue Type: Bug
          Components: API / DataStream
    Affects Versions: 1.15.2
            Reporter: Zakelly Lan
             Fix For: 1.16.0
         Attachments: image-2022-09-28-14-32-28-116.png, image-2022-09-28-14-35-47-954.png

I notice that the key group range in consumer side of Kafka Shuffle is not aligned with the producer side, there are two problems:
 # The data partitioning of the sink(producer) is exactly the same way as a keyed stream that as the same maximum parallelism as the number of kafka partitions does, but in consumer side the number of partitions and key groups are not the same.
 # There is a distribution of assigning kafka partitions to consumer subtasks (See KafkaTopicPartitionAssigner#assign), but the producer of Kafka Shuffle simply assume the partition index equals the subtask index. e.g. !image-2022-09-28-14-32-28-116.png|width=1133,height=274!

My proposed change:
 # Set the max parallelism of the key stream in consumer size as the number of kafka partitions. 
 # Use the same method when assigning kafka partitions to consumer subtasks to maintain a map from subtasks to kafka partitions, which is used by the producer to insert into the right partition for data for a subtask. i.e. !image-2022-09-28-14-35-47-954.png|width=1030,height=283!



--
This message was sent by Atlassian Jira
(v8.20.10#820010)