我正在尝试使用 Kafka Connect 从 Oracle 数据库中提取数据。 Kafka 连接器提供的默认对象是“GenericRecord”类型。这使得它过于具体，并导致通过执行 record.getAsString("someIDENTIFIER") 来获取数据的情况。是否有可能获得特定类型的对象而不是 GenericRecord 类型。
【问题讨论】：
标签： apache-kafka apache-kafka-connect
Kafka Connect 源连接器使用 SourceRecord 对象，并且 Kafka Connect 工作器被配置为使用 converter 将 SourceRecord 序列化为二进制形式，然后写入 Kafka 主题. Kafka Connect 附带一个 JSON 转换器，而 Confluent 提供一个 Avro 转换器。因此，写入 Kafka 的消息的二进制形式取决于您使用的转换器。
（同样，接收器连接器使用SinkRecord 对象，Kafka Connect 工作程序使用其转换器将从 Kafka 读取的消息的二进制形式反序列化为连接器处理的SinkRecord 对象。）
听起来您正在编写一个 Kafka 消费者并在那里看到 GenericRecord 对象。如果是这样，那么您可能已将 Kafka Connect 工作程序配置为使用 Confluent 的 Avro 转换器，该转换器对于 JDBC 连接器等源连接器将 SourceRecord 转换为 Kafka Connect 然后写入 Kafka 主题的 Avro 二进制格式。然后，您的客户端可能会使用配置了 Avro 反序列化器的 Kafka 消费者，除非您为反序列化器提供 Avro 模式以使用它，否则它将反序列化 Avro 编码消息到 Avro GenericRecord。
但是，您可以将应用程序配置为了解特定版本的 Avro 架构，并让构建系统为该版本的 Avro 架构生成代码，以创建将反序列化 Avro 编码消息的特定代码转换为模式描述的内存形式。在 Java 中，这意味着您将从模式生成类，然后在代码中使用生成的类将 GenericRecord 复制到类的实例中。请参阅this complete consumer example，特别是this line 以了解从GenericRecord 的转换。在该示例中，LogLine 是从 Avro 模式生成的类：
GenericRecord genericEvent = (GenericRecord) messageAndMetadata.message();
LogLine event = (LogLine) SpecificData.get().deepCopy(LogLine.SCHEMA$, genericEvent);
Avro 的一个显着优势是它直接支持模式演化，而 Confluent 的模式注册表利用了这一点。因此，尽管源连接器可能会演变为表生成的 Avro 模式以响应数据库中表结构的变化，但只要更改数据库模式以使 Avro 模式向后和/或向前兼容，Avro 库您的客户端应用程序使用的消息将从 Avro 架构自动转换为您的应用程序使用的 Avro 架构。
当然，有时您会更改应用程序以使用新的 Avro 架构，但这不必同时进行。事实上，如果您将 Schema Registry 配置为强制模式版本向前和向后兼容，您可以更改您的客户端应用程序before 或 after 数据库更改并且JDBC 源连接器开始使用新版本的 Avro 架构。
【讨论】：