Kafka 主题 Avro 键未从 Cassandra 源连接器填充

Question

我有一个 Kafka 主题，它具有以下 Avro Key Schema：

{
  "type": "record",
  "name": "onboardingStatusKey",
  "namespace": "onboarding",
  "fields": [
    {
      "name": "time_created",
      "type": "long",
      "logicalType": "timestamp-millis"
    },
    {
      "name": "local_enterprise_id",
      "type": "string"
    },
    {
      "name": "source_request_reference",
      "type": "string"
    },
    {
      "name": "onboarding_request_id",
      "type": "string"
    },
    {
      "name": "system_code",
      "type": "string"
    }
  ],
  "version": "1"
}

我在 Cassandra 中还有一个表，其中包含这些字段（以及其他一些字段）。我正在寻找一个源连接器，它将数据从 Cassandra 表移动到我的 Kafka 主题。

我的源连接器配置是：

{
    "name": "cassandra-source-{{TableName}}-3",
    "config": {
        "connector.class": "com.datamountaineer.streamreactor.connect.cassandra.source.CassandraSourceConnector",
        "tasks.max": "2",
        "connect.cassandra.contact.points": "{{CassandraContactPoints}}",
        "connect.cassandra.port": "{{CassandraContactPort}}",
        "connect.cassandra.ssl.enabled": "true",
        "connect.cassandra.trust.store.path": "/trusted-clusters.jks",
        "connect.cassandra.trust.store.password": "{{TruststorePWD}}",
        "connect.cassandra.trust.store.type": "JKS",
        "connect.cassandra.username": "{{CassandraUsername}}",
        "connect.cassandra.password": "{{CassandraPassword}}",
        "connect.cassandra.key.space": "{{Keyspace}}",

        "connect.cassandra.import.mode": "incremental",
        "connect.cassandra.kcql": "INSERT INTO {{TopicName}} SELECT * FROM {{TableName}} PK time_created INCREMENTALMODE=TIMESTAMP WITHKEY(time_created, local_enterprise_id, source_request_reference, onboarding_request_id, system_code)",

        "key.converter": "io.confluent.connect.avro.AvroConverter",
        "key.converter.schema.registry.url": "https://{{SchemaUsername}}:{{SchemaPassword}}@kafka-schema.{{SchemaURL}}.cnodes.io:8085",
        "key.converter.basic.auth.credentials.source": "URL",

        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "value.converter.schema.registry.url": "https://{{SchemaUsername}}:{{SchemaPassword}}@kafka-schema.{{SchemaURL}}.cnodes.io:8085",
        "value.converter.basic.auth.credentials.source": "URL",

        "errors.tolerance": "all",
        "errors.log.include.messages": true,
        "errors.log.enable": true
    }
}

表中的行正在发送到主题，但每条记录中缺少键（仅出现值部分）

我缺少什么？

Answer 1

免责声明：我没有使用此连接器的经验

您似乎缺少 WITHKEY 查询的一部分，用于选择一对多字段作为记录键写入。

请记住，源连接器创建自己的 Avro 架构，并且不会使用您在兼容性检查之外预定义的架构（如果启用）