TiCDC发送到kafka的消息分区参数不生效

🪐 TiDB 技术问题
1

【TiDB 使用环境】测试
启动命令中设置的分区数量是3，但配置文件中动态生成的topic的分区数量并没有按照命令行上的分区数量设置。
./cdc cli changefeed create --server=http://`````````:8300 --sink-uri=“kafka://`````````:9092/topic_ticdc_saleslink_uc_employee?protocol=canal-json&kafka-version=2.6.2&partition-num=3&max-message-bytes=67108864&replication-factor=3” --changefeed-id=“data-syc-two-way-saleslink-uc-employee” --config=/data/tidb/ticdc-changefeed/changefeed-data-syc-saleslink_uc_employee.conf

[consistent]
level = "none"

[filter]
rules = ['saleslink_uc_employee.emp_employee']

[[filter.event-filters]]
matcher = ['saleslink_uc_employee.emp_employee']

[mounter]
worker-num = 16

# Sink 配置
[sink]
protocol = "debezium"
dispatchers = [
  {matcher = ['saleslink_uc_employee.emp_employee'], topic = "topic_ticdc_{schema}_{table}", partition = "index-value"}
]

结果默认topic是3分区topic_ticdc_saleslink_uc_employee，配置内表相关的topic用的是默认12分区

[root@saleslink-uat-ticdc-01 bin]# ./kafka-topics.sh --describe --bootstrap-server *******:9092 --topic topic_ticdc_saleslink_uc_employee
Topic: topic_ticdc_saleslink_uc_employee        PartitionCount: 3       ReplicationFactor: 1    Configs: 
        Topic: topic_ticdc_saleslink_uc_employee        Partition: 0    Leader: 103     Replicas: 103   Isr: 103        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee        Partition: 1    Leader: 101     Replicas: 101   Isr: 101        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee        Partition: 2    Leader: 102     Replicas: 102   Isr: 102        Elr: N/A        LastKnownElr: N/A

[root@saleslink-uat-ticdc-01 bin]# ./kafka-topics.sh --describe --bootstrap-server *******:9092 --topic topic_ticdc_saleslink_uc_employee_emp_employee
Topic: topic_ticdc_saleslink_uc_employee_emp_employee   PartitionCount: 12      ReplicationFactor: 1    Configs: 
        Topic: topic_ticdc_saleslink_uc_employee_emp_employee   Partition: 0    Leader: 101     Replicas: 101   Isr: 101        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee_emp_employee   Partition: 1    Leader: 102     Replicas: 102   Isr: 102        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee_emp_employee   Partition: 2    Leader: 103     Replicas: 103   Isr: 103        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee_emp_employee   Partition: 3    Leader: 101     Replicas: 101   Isr: 101        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee_emp_employee   Partition: 4    Leader: 102     Replicas: 102   Isr: 102        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee_emp_employee   Partition: 5    Leader: 103     Replicas: 103   Isr: 103        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee_emp_employee   Partition: 6    Leader: 101     Replicas: 101   Isr: 101        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee_emp_employee   Partition: 7    Leader: 102     Replicas: 102   Isr: 102        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee_emp_employee   Partition: 8    Leader: 103     Replicas: 103   Isr: 103        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee_emp_employee   Partition: 9    Leader: 101     Replicas: 101   Isr: 101        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee_emp_employee   Partition: 10   Leader: 102     Replicas: 102   Isr: 102        Elr: N/A        LastKnownElr: N/A
        Topic: topic_ticdc_saleslink_uc_employee_emp_employee   Partition: 11   Leader: 103     Replicas: 103   Isr: 103        Elr: N/A        LastKnownElr: N/A
2

感觉像BUG。没找到可以指定动态生成的topic的分区参数…

3

[sink]
protocol = “debezium”
dispatchers = [
{matcher = [‘saleslink_uc_employee.emp_employee’], topic = “topic_ticdc_{schema}_{table}”, partition = “3”}
]

4

./kafka-topics.sh --create --bootstrap-server :9092 --topic topic_ticdc_saleslink_uc_employee_emp_employee --partitions 3 --replication-factor 3

5

分区数只能扩不能缩，我猜你这个topic是事先手动创建的12分区。

6

我试试，没在文档上看到partitions参数，只有partition-num 。还有配置文件里面的partition 应该是指定分发到哪个分区规则的吧
https://docs.pingcap.com/zh/tidb/stable/ticdc-sink-to-kafka/#partition-分发器

7

是不是你topic没有指定 partition-num ？

8

我启动命令里面 partition-num=3

10

https://docs.pingcap.com/zh/tidb/stable/ticdc-sink-to-kafka/#partition-分发器

我看文档里面是支持动态发送到partition的呀，这个功能没支持吗？

11

我看错了，如果下游没有分区，则会主动创建分区，如果已经存在了就不会再创建了，这里是不是提前手动创建了分区呢

12

如果事先手动创建了分区，会有什么影响吗

13

那数据发送就会往12个分区都发送

14

分区数量是多少

15

是不是之前kafka的topic已经存在分区了

16

配置里面的分区数量是多少呢

17

:point_right: partition-num=3 只对“默认 topic”生效
:point_right: 你配置的 topic_ticdc_{schema}_{table} 是 动态 topic ，不会使用这个参数
:point_right: 动态 topic 的分区数 = Kafka broker 默认分区

2 个赞
18

:white_check_mark: 方案1（推荐）：提前建好 topic

./kafka-topics.sh --create
–bootstrap-server xxx:9092
–topic topic_ticdc_saleslink_uc_employee_emp_employee
–partitions 3
–replication-factor 3

:point_right: 优点：

  • 完全可控
  • 最稳定
  • 生产推荐

:white_check_mark: 方案2：关闭 Kafka 自动创建 topic

在 Kafka 配置中：

auto.create.topics.enable=false

:point_right: 然后：

  • TiCDC 写入失败（因为 topic 不存在）
  • 强制你必须提前创建

:white_check_mark: 方案3：修改 Kafka 默认分区数

num.partitions=3

:point_right: 缺点：

  • 影响所有新 topic（不推荐生产）
2 个赞
19

我也遇到了，没有解决。后续自己测试看看