[
https://issues.apache.org/jira/browse/FLINK-26033?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17514050#comment-17514050
]
shizhengchao commented on FLINK-26033:
--------------------------------------
master: 2049f849a130e738759001c5a6b9d85834da08d0
1.13: 8b18e74cec71c9c480bb747fb2f77a55f675ca2a
1.14: d4f374237b21aab93f5d185cb2d1e15a29c9c537
> In KafkaConnector, when 'sink.partitioner' is configured as 'round-robin', it
> does not take effect
> --------------------------------------------------------------------------------------------------
>
> Key: FLINK-26033
> URL: https://issues.apache.org/jira/browse/FLINK-26033
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: 1.13.3, 1.11.6, 1.12.7, 1.14.3
> Reporter: shizhengchao
> Assignee: shizhengchao
> Priority: Major
> Labels: pull-request-available
>
> In KafkaConnector, when 'sink.partitioner' is configured as 'round-robin', it
> does not take effect. Flink treats 'default' and 'round-robin' as the same
> strategy.
> {code:java}
> //代码占位符
> public static Optional<FlinkKafkaPartitioner<RowData>>
> getFlinkKafkaPartitioner(
> ReadableConfig tableOptions, ClassLoader classLoader) {
> return tableOptions
> .getOptional(SINK_PARTITIONER)
> .flatMap(
> (String partitioner) -> {
> switch (partitioner) {
> case SINK_PARTITIONER_VALUE_FIXED:
> return Optional.of(new
> FlinkFixedPartitioner<>());
> case SINK_PARTITIONER_VALUE_DEFAULT:
> case SINK_PARTITIONER_VALUE_ROUND_ROBIN:
> return Optional.empty();
> // Default fallback to full class name of the
> partitioner.
> default:
> return Optional.of(
> initializePartitioner(partitioner,
> classLoader));
> }
> });
> } {code}
> They both use kafka's default partitioner, but the actual There are two
> scenarios for the partition on DefaultPartitioner:
> 1. Random when there is no key
> 2. When there is a key, take the modulo according to the key
> {code:java}
> // org.apache.kafka.clients.producer.internals.DefaultPartitioner
> public int partition(String topic, Object key, byte[] keyBytes, Object value,
> byte[] valueBytes, Cluster cluster) {
> if (keyBytes == null) {
> // Random when there is no key
> return stickyPartitionCache.partition(topic, cluster);
> }
> List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
> int numPartitions = partitions.size();
> // hash the keyBytes to choose a partition
> return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
> } {code}
> Therefore, KafkaConnector does not have a round-robin strategy.But we can
> borrow from kafka's RoundRobinPartitioner
> {code:java}
> //代码占位符
> public class RoundRobinPartitioner implements Partitioner {
> private final ConcurrentMap<String, AtomicInteger> topicCounterMap = new
> ConcurrentHashMap<>();
> public void configure(Map<String, ?> configs) {}
> /**
> * Compute the partition for the given record.
> *
> * @param topic The topic name
> * @param key The key to partition on (or null if no key)
> * @param keyBytes serialized key to partition on (or null if no key)
> * @param value The value to partition on or null
> * @param valueBytes serialized value to partition on or null
> * @param cluster The current cluster metadata
> */
> @Override
> public int partition(String topic, Object key, byte[] keyBytes, Object
> value, byte[] valueBytes, Cluster cluster) {
> List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
> int numPartitions = partitions.size();
> int nextValue = nextValue(topic);
> List<PartitionInfo> availablePartitions =
> cluster.availablePartitionsForTopic(topic);
> if (!availablePartitions.isEmpty()) {
> int part = Utils.toPositive(nextValue) %
> availablePartitions.size();
> return availablePartitions.get(part).partition();
> } else {
> // no partitions are available, give a non-available partition
> return Utils.toPositive(nextValue) % numPartitions;
> }
> }
> private int nextValue(String topic) {
> AtomicInteger counter = topicCounterMap.computeIfAbsent(topic, k -> {
> return new AtomicInteger(0);
> });
> return counter.getAndIncrement();
> }
> public void close() {}
> } {code}
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
