Kafka Connect 忽略指定的主题策略

Question

我想使用以下连接器配置将多个表数据发布到同一个 Kafka 主题，但我看到以下异常

例外

原因：io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException：正在注册的模式与早期的模式不兼容；错误代码：409

连接器似乎忽略了主题策略属性集并继续使用旧的 ${topic}-key 和 ${topic}-value 主题。

[2019-04-25 22:43:45,590] INFO AvroConverterConfig values: 
    schema.registry.url = [http://schema-registry:8081]
    basic.auth.user.info = [hidden]
    auto.register.schemas = true
    max.schemas.per.subject = 1000
    basic.auth.credentials.source = URL
    schema.registry.basic.auth.user.info = [hidden]
    value.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
    key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy

连接器配置

    curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{
      "name": "two-in-one-connector",
      "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "xxxxxxx",
        "database.port": "3306",
        "database.user": "xxxxxxx",
        "database.password": "xxxxxxxxx",
        "database.server.id": "18405457",
        "database.server.name": "xxxxxxxxxx",
        "table.whitelist": "customers,phone_book",
        "database.history.kafka.bootstrap.servers": "broker:9092",
        "database.history.kafka.topic": "dbhistory.customer",
        "transforms": "dropPrefix",
        "transforms.dropPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.dropPrefix.regex":"(.*)",
        "transforms.dropPrefix.replacement":"customer",
        "key.converter.key.subject.name.strategy": "io.confluent.kafka.serializers.subject.TopicRecordNameStrategy",
        "value.converter.value.subject.name.strategy": "io.confluent.kafka.serializers.subject.TopicRecordNameStrategy"
      }
    }'

Answer 1

尝试将策略类设置为连接器配置 (JSON) 文件中的以下参数，而不是“key.converter.key.subject.name.strategy”和“value.converter.value .subject.name.strategy"

"key.subject.name.strategy" "value.subject.name.strategy"