[
https://issues.apache.org/jira/browse/FLINK-29437?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17611494#comment-17611494
]
Zakelly Lan commented on FLINK-29437:
-------------------------------------
After offline sync with [~ym] , we agree to fix this as I proposed.
> 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
> Assignee: Zakelly Lan
> Priority: Major
> 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 side 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)
