无法从 debezium-postgres 的 kafka-stream 读取 kafka-stream 数据答案

【问题标题】：Unable to read kafka-stream data from debezium-postgres's kafka-stream无法从 debezium-postgres 的 kafka-stream 读取 kafka-stream 数据
【发布时间】：2019-03-16 03:40:35
【问题描述】：

我使用以下命令启动了 kafka 连接器：

./bin/connect-standalone etc/schema-registry/connect-avro-standalone.properties etc/kafka-connect-postgres/connect-postgres.properties

connect-avro-standalone.properties 中的序列化道具是：

key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081

internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

我创建了一个 java 后端，它监听这个 kafka 流主题，并且能够通过每次添加/更新/删除从 postgres 获取数据。
但是数据以某种未知的编码格式出现，这就是为什么我无法正确读取数据的原因。
下面是相关代码sn-p：

properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
Serdes.String().getClass().getName());
properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
Serdes.ByteArray().getClass().getName());

StreamsBuilder streamsBuilder = new StreamsBuilder();

final Serde<String> stringSerde = Serdes.String();
final Serde<byte[]> byteSerde = Serdes.ByteArray();

streamsBuilder.stream(Pattern.compile(getTopic()), Consumed.with(stringSerde, byteSerde))
.mapValues(data -> {
  System.out.println("->"+new String(data));
  return data;
});

我对需要更改的位置和内容感到困惑；在 avro 连接器属性或 java 端代码中

【问题讨论】：

标签： avro apache-kafka-streams apache-kafka-connect debezium

【解决方案1】：

您的 Kafka Connect 配置意味着 Kafka 主题上的消息将被 Avro 序列化：

value.converter=io.confluent.connect.avro.AvroConverter

这意味着您需要在 Streams 应用中使用 Avro 反序列化。更多详情请看这里：https://docs.confluent.io/current/streams/developer-guide/datatypes.html#avro

【讨论】：

我可以把它改成 value.converter=org.apache.kafka.connect.json.JsonConverter
是的，如果您更愿意使用 JSON，您也可以这样做。
只是一个问题....在选择 JsonConverter 时会不会对性能产生影响？
不能真正评论编码吞吐量方面的性能，但可以肯定的是，Avro 在传输的字节数和主题大小方面更有效。请记住，您需要部署架构注册表才能使用 Avro。顺便提一句。 here 是关于配置 Avro 序列化程序的 Debezium 文档。