将 Kafka 连接结构转换为 POJO

Question

我正在开发从数据库捕获数据更改 (CDC) 的组件，对数据应用一些业务逻辑，然后将它们发送到一个新主题，该主题是一个天蓝色的事件中心主题。

我的堆栈包括：

• 骆驼管道
• 嵌入式 Debezium（用于 CDC）

这个来自 debezium 的 article 是我开始工作的基础。 Converter 用于将 Struct 转换为 POJO。

  @Converter
  public static class Converters {

    @Converter
    public static Question questionFromStruct(Struct struct) {                   
      return new Question(struct.getInt64("id"), struct.getString("text"),
          struct.getString("email"));
    }

    @Converter
    public static Answer answerFromStruct(Struct struct) {                       
      return new Answer(struct.getInt64("id"), struct.getString("text"),
          struct.getString("email"), struct.getInt64("question_id"));
    }
  }

就我而言，我正在处理的 POJO 是 Avro 生成的。它还包括 20 多个我不想手动设置的属性。

我考虑过 MapStruct 或 Dozer 等映射器。但是不处理这种映射。

关于如何更自动化地处理struct 有什么见解吗？

Answer 1

如 Debezium 引擎文档中所述

在内部，引擎使用适当的 Kafka Connect 转换器实现，转换被委托给它

这意味着，您需要反转 Converter 方法，然后使用 Kafka Deserializer 类将数据返回到最初生成的 POJO。

---

假设您使用了 Confluent 的 AvroConverter，那么您应该能够使用 AvroConverter#fromConnectData 返回 byte[] 作为键/值。不过，您首先需要为您的Struct 使用 Connect Schema 实例。

从那里，您可以将字节传递给KafkaAvroDeserializer#deserialize 并将响应转换为您生成的类。

---

另外，在 Debezium/Kafka Connect 中转换数据的正确方法是使用 Simple-Message-Transforms 并保留在 Struct/Schema API 中。