SOLVED canal无法写入kafka多分区

[navyaijm2017]

[junyu-cloudcanal]

@navyaijm2017 在 canal无法写入kafka多分区 中说：

已解决，正确的配置

mq config

canal.mq.dynamicTopic=db_chj_test.sbtest2
#canal.mq.partition=3
canal.mq.partitionsNum=3
canal.mq.partitionHash=db_chj_test.sbtest2:id

补充下。

canal.mq.partition 这个参数实际上表示的是在 canal.mq.partitionsNum 和 canal.mq.partitionHash 未设置的情况下，默认写入的partition。以下是参考源码

 for (FlatMessage flatMessage : flatMessages) {
                /**canal.mq.partitionsNum和canal.mq.partitionHash **/
                if (mqDestination.getPartitionHash() != null && !mqDestination.getPartitionHash().isEmpty()) {
                    FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
                        mqDestination.getPartitionsNum(),
                        mqDestination.getPartitionHash(),
                        this.mqProperties.isDatabaseHash());
                    int length = partitionFlatMessage.length;
                    for (int i = 0; i < length; i++) {
                        FlatMessage flatMessagePart = partitionFlatMessage[i];
                        if (flatMessagePart != null) {
                            records.add(new ProducerRecord<>(topicName, i, null, JSON.toJSONBytes(flatMessagePart,
                                SerializerFeature.WriteMapNullValue)));
                        }
                    }
                } else {
                    /**canal.mq.partition**/
                    final int partition = mqDestination.getPartition() != null ? mqDestination.getPartition() : 0;
                    records.add(new ProducerRecord<>(topicName, partition, null, JSON.toJSONBytes(flatMessage,
                        SerializerFeature.WriteMapNullValue)));
                }
            }

[navyaijm2017]

已解决，正确的配置

mq config

canal.mq.dynamicTopic=db_chj_test.sbtest2
#canal.mq.partition=3
canal.mq.partitionsNum=3
canal.mq.partitionHash=db_chj_test.sbtest2:id