You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Yuan Mei (Jira)" <ji...@apache.org> on 2022/11/21 12:56:00 UTC
[jira] [Resolved] (FLINK-29437) The partition of data before and after the Kafka Shuffle are not aligned
[ https://issues.apache.org/jira/browse/FLINK-29437?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Yuan Mei resolved FLINK-29437.
------------------------------
Resolution: Fixed
merged commit [{{0b81668}}|https://github.com/apache/flink/commit/0b816680e51283d261eb7b7bce560dd1641691ce] into apache:master
> 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, Connectors / Kafka
> Affects Versions: 1.15.3
> Reporter: Zakelly Lan
> Assignee: Zakelly Lan
> Priority: Major
> Labels: pull-request-available
> Fix For: 1.17.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 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)