TiCDC同步pursal消息topic转发规则失效以及msgKey问题

🪐 TiDB 技术问题 备份&数据迁移
1 
[filter]
rules = ['test.test1','test.test2']
[sink]
dispatchers = [
  {matcher = ['test.test1','test.test2'], topic = "{schema}_{table}", partition = "index-value"}
]

/cdc cli changefeed create \
    --server=http://127.0.0.1:8301 \
--sink-uri="pulsar+http://[ip]:[port]/persistent://pulsar-jegrg79egk7km7/test/test_test1?protocol=canal-json" \
--config=./test_changefeed.toml \
--changefeed-id="test-sync-task"

我想问下这个topic的分发规则为什么没有生效,我已经在pulsar服务器创建了两个topic分别为[test_test1,test_test2] 然后指定默认的topic为test_test1 经过测试发现[filter]过滤的规则rules生效了 但是[sink]下的dispatchers没有生效 无论改test1还是test2都会发到默认的test_test1这个topic中 而且消息的key都是[库名.表名] 而不是我指定的index-value 是这个配置没有用吗

2

你这个配置是过滤 t3_test 库下面的 test1，和 test2 两个表。
下面 matchser 的是 test 库下的 test1,test2 两个表。为啥这么配置？

你现在写下游 kafka 的数据都是 t3_test 库下的 test1/test2 两个表数据吧。

3

不好意思是test库 写帖子的时候改错了 已经纠正 我的下游是pulsar消息队列

4

看下日志呢，有没有相关报错

5 
{
    "upstream_id": 7405574986668705397,
    "namespace": "default",
    "id": "test-sync-task",
    "sink_uri": "pulsar+http://[ip]:[port]/persistent://pulsar-jegrg79egk7km7/test/test_test1?protocol=canal-json",
    "create_time": "2025-03-19T15:26:51.495825464+08:00",
    "start_ts": 456751634519687175,
    "config": {
        "memory_quota": 1073741824,
        "case_sensitive": false,
        "force_replicate": false,
        "ignore_ineligible_table": false,
        "check_gc_safe_point": true,
        "enable_sync_point": false,
        "enable_table_monitor": false,
        "bdr_mode": false,
        "sync_point_interval": 600000000000,
        "sync_point_retention": 86400000000000,
        "filter": {
            "rules": [
                "test.test1",
                "test.test2"
            ]
        },
        "mounter": {
            "worker_num": 16
        },
        "sink": {
            "protocol": "canal-json",
            "csv": {
                "delimiter": ",",
                "quote": "\"",
                "null": "\\N",
                "include_commit_ts": false,
                "binary_encoding_method": "base64",
                "output_old_value": false,
                "output_handle_key": false
            },
            "dispatchers": [
                {
                    "matcher": [
                        "test.test2"
                    ],
                    "topic": "test_test2"
                },
                {
                    "matcher": [
                        "test.test1"
                    ],
                    "partition": "index-value",
                    "topic": "test_test1"
                }
            ],
            "encoder_concurrency": 32,
            "terminator": "\r\n",
            "date_separator": "day",
            "enable_partition_separator": true,
            "enable_kafka_sink_v2": false,
            "only_output_updated_columns": false,
            "delete_only_output_handle_key_columns": false,
            "content_compatible": false,
            "pulsar_config": {
                "pulsar-producer-cache-size": 10240,
                "authentication-token": "***",
                "connection-timeout": 15,
                "operation-timeout": 30,
                "batching-max-messages": 1000,
                "batching-max-publish-delay": 10,
                "send-timeout": 30
            },
            "advance_timeout": 150,
            "send_bootstrap_interval_in_sec": 120,
            "send_bootstrap_in_msg_count": 10000,
            "send_bootstrap_to_all_partition": true,
            "send-all-bootstrap-at-start": false,
            "debezium_disable_schema": false,
            "debezium": {
                "output_old_value": true
            },
            "open": {
                "output_old_value": true
            }
        },
        "consistent": {
            "level": "none",
            "max_log_size": 64,
            "flush_interval": 2000,
            "meta_flush_interval": 200,
            "encoding_worker_num": 16,
            "flush_worker_num": 8,
            "use_file_backend": false,
            "memory_usage": {
                "memory_quota_percentage": 50
            }
        },
        "scheduler": {
            "enable_table_across_nodes": false,
            "region_threshold": 100000,
            "write_key_threshold": 0
        },
        "integrity": {
            "integrity_check_level": "none",
            "corruption_handle_level": "warn"
        },
        "changefeed_error_stuck_duration": 1800000000000,
        "synced_status": {
            "synced_check_interval": 300,
            "checkpoint_interval": 15
        }
    },
    "state": "normal",
    "creator_version": "v8.5.1",
    "resolved_ts": 456751634519687175,
    "checkpoint_ts": 456751634519687175,
    "checkpoint_time": "2025-03-19 15:26:51.272"
}

任务启动后没报错 就是正常打印了这些配置

6

是执行的 DDL 和 DML 都出现了这个情况吗？从 config 看 test.test2 的应该写到 test_test2 才对，发一下日志看看呢