Kafka 连接 - JDBC 自定义 Avro 模式答案

【问题标题】：Kafka connect - JDBC custom Avro schemaKafka 连接 - JDBC 自定义 Avro 模式
【发布时间】：2020-04-09 08:48:13
【问题描述】：

我正在关注关于 kafka 连接的教程，我想知道是否有可能接收到某种类型的消息。

教程：https://www.confluent.io/blog/simplest-useful-kafka-connect-data-pipeline-world-thereabouts-part-1/

与教程中的表格一样，架构如下所示：

{
   "namespace": "avro",
   "type": "record",
   "name": "Audit",
   "fields": [
      {"name": "c1", "type": "int"},
      {"name": "c2", "type": "string"},
      {"name": "create_ts", "type": "long"},
      {"name": "update_ts", "type": "long"}
   ]
}

基于 avro 格式，我使用 maven 生成了一个类。

然后我用我的类型定义了消费者工厂：

public ConsumerFactory<String, Audit> auditConsumerFactory() { ... )

还有 KafkaListener：

@KafkaListener(topics = "${kafka.mysql.topic}", containerFactory =   "mysqlKafkaListenerContainerFactory")
public void receive(Audit audit) {
     System.out.println(audit);
     this.latch.countDown();
}

但最后我得到了这样的错误：

2019-12-16 21:56:50.139 ERROR 31862 --- [ntainer#0-0-C-1] o.s.kafka.listener.LoggingErrorHandler   : Error while processing: null
org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition mysql-audit-0 at offset 4. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 1
Caused by: org.apache.kafka.common.errors.SerializationException: Could not find class audit specified in writer's schema whilst finding reader's schema for a SpecificRecord.

编辑 ConsumerFactory 与反序列化器：

    public ConsumerFactory<String, Audit> auditConsumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfiguration.getKafkaBootstrapAddress());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        props.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
        props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
        return new DefaultKafkaConsumerFactory(props);
    }

审计.avsc

{
  "type": "record",
  "name": "avro.Audit",
  "fields": [
    {
      "name": "c1",
      "type": "int"
    },
    {
      "name": "c2",
      "type": "string"
    },
    {
      "name": "create_ts",
      "type": {
        "type": "long",
        "connect.version": 1,
        "connect.name": "org.apache.kafka.connect.data.Timestamp",
        "logicalType": "timestamp-millis"
      }
    },
    {
      "name": "update_ts",
      "type": {
        "type": "long",
        "connect.version": 1,
        "connect.name": "org.apache.kafka.connect.data.Timestamp",
        "logicalType": "timestamp-millis"
      }
    }
  ],
  "connect.name": "avro.Audit"
}

我在Github找到了我的问题的答案

【问题讨论】：

Could not find class audit ... 您的 Avro 记录或 Java 类的名称不是小写的，那么为什么 Avro 会认为是小写呢？另外，请显示您的 Kafka Connect 和 Spring 反序列化器属性
您还可以从/schemas/ids/1 查看注册表中的架构（假设它实际上是 ID 1，如错误所示）
嗨，我把与 Audit 类相关的所有内容都大写了。我已经检查了模式注册表上的版本。将控制中心中的一个与我的主题和本地的一个对齐。仍然出现同样的错误
您显示的架构是注册表中的架构吗？
我现在发现问题可能与 Audit 类所在的位置有关。我有一个包 avro.Audit 未在模式注册表中定义。我认为这就是问题所在。是：“名称”：“审计”，应该是“名称”：“avro.Audit”

标签： apache-kafka avro spring-kafka apache-kafka-connect confluent-schema-registry

【解决方案1】：

我不知道是否有关于这个问题的另一个线程，但最后 Confluence 解决了这个问题。将这三行添加到 JDBC 连接器

"transforms": "AddNamespace", "transforms.AddNamespace.type": "org.apache.kafka.connect.transforms.SetSchemaMetadata$Value", "transforms.AddNamespace.schema.name": "my.namespace.NameOfTheSchema",

KAFKA-7883

【讨论】：