Liam created FLINK-31762:
----------------------------
Summary: Subscribe 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
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}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
