从 Kafka 主题反序列化 FlinkKafkaProducer 的事件导致 JSON 条记录为空

Question

编辑 我准备了最小的可复制示例 https://github.com/kazuhirokomoda/flink-kafka-druid

我有一个数据管道和代码，与我之前的问题中提到的完全相同（除非在此 post 底部另有说明）： 基本上我使用的是 Jackson 的 ObjectMapper 在自定义 KafkaSerializationSchema（即 ExampleDataSerializationSchema）中序列化事件。

出于测试目的，我在裸机服务器上准备了 Kafka 2.6，遵循 https://kafka.apache.org/quickstart

我使用 kafka-console-consumer.sh 检查从 Flink 作业发送到 Kafka 主题的内容，但我只看到空的 JSON 记录，而我期待的是非空记录。

bin/kafka-console-consumer.sh --topic <topic> --from-beginning --bootstrap-server localhost:9092
...
{}
{}
{}
^CProcessed a total of 287 messages

备注

• 毫不奇怪，按照 The Kafka quickstart 从第 1 步到第 5 步：阅读事件（i.e.string input/output 使用 Kafka 控制台 producer/consumer）工作正常。
• 从 Flink 作业发送（固定）字符串（而不是 Jackson/JSON）也工作正常。 （见下面的代码）

bin/kafka-console-consumer.sh --topic <topic> --from-beginning --bootstrap-server localhost:9092
...
fixed string to send to kafka
fixed string to send to kafka
fixed string to send to kafka
^CProcessed a total of 287 messages

问题

• 当使用自定义序列化器时（即Flink的ExampleDataSerializationSchema，如我之前的问题所示），我是否也需要定义自定义de序列化器？
  • 如果是，我该如何实现它，我应该把它放在哪里，以便 Kafka 控制台消费者（最终是德鲁伊，这是我上一个问题的目标）可以使用它？
• 记录为空的原因是序列化还是反序列化，还是两者兼而有之？
• 就(JSON)serialization/deserialization而言，Kafka 0.11.x和2.x有什么区别需要注意吗？
• 如果我想 serialize/deserialize 事件作为 JSON，是否有 Jackson 的替代方案？

再次感谢您的帮助。

代码

Flink -> Kafka

object KafkaSink {
  def sendToKafka(exampleDataStream: DataStream[ExampleData], props: ParameterTool): DataStreamSink[ExampleData] = {
    // defined in my previous question
  }

  // just for testing
  def sendToKafkaString(exampleDataStream: DataStream[ExampleData], props: ParameterTool): DataStreamSink[String] = {
    val topic: String = ...
    val properties: Properties = ... 

    val producer = new FlinkKafkaProducer[String](
      topic,
      new ExampleDataStringSerializationSchema(topic),
      properties,
      FlinkKafkaProducer.Semantic.EXACTLY_ONCE)

    val exampleDataStreamString = exampleDataStream.map(_ => "fixed string to send to kafka")
    exampleDataStreamString.addSink(producer)
  }
}

import java.nio.charset.StandardCharsets

class ExampleDataStringSerializationSchema(topic: String) extends KafkaSerializationSchema[String]{
  override def serialize(element: String, timestamp: java.lang.Long): ProducerRecord[Array[Byte], Array[Byte]] = {
    new ProducerRecord[Array[Byte], Array[Byte]](topic, element.getBytes(StandardCharsets.UTF_8))
  }
}

Answer 1

ObjectMapper 包装器解决方法对我有用。

https://github.com/kazuhirokomoda/flink-kafka-druid/pull/1

谢谢onecricketeer！