反序列化 Avro 序列化 Kafka 流的问题答案

【问题标题】：Problems deserializing an Avro serialized Kafka Stream反序列化 Avro 序列化 Kafka 流的问题
【发布时间】：2018-08-11 02:49:27
【问题描述】：

我在尝试实现商店时收到异常。我正在运行 Kafka 1.0、Confluent 的 Schema Registry 4.0 和 Avro 1.8.2。我已经使用 Avro 的 maven 插件生成了 Pojo，并使用 Confluent maven 插件将模式部署到了 Confluent 服务器。我能够向 STREAM1 主题发送消息。这是设置流的代码：

Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-pipe");
properties.put(StreamsConfig.CLIENT_ID_CONFIG, "cleant-id");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "http://localhost:9092");
properties.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
StreamsBuilder builder = new StreamsBuilder();

Serde<T> pojoSerde = new SpecificAvroSerde<>();
final Map<String, String> serdeConfig = Collections.singletonMap(
        AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
pojoSerde.configure(serdeConfig, false);

Consumed<String, Pojo> consumed = Consumed.with(Serdes.String(), pojoSerde);
KStream<String, Pojo> source = builder.stream(TopicName.STREAM1.toString(), consumed);
KTable<String, Long> storePojoCount = source
        .groupBy((key, value) -> key)
        .count(Materialized.as(StoreName.STORE_WORD_COUNT.toString()));

Produced<String, Long> produced = Produced.with(Serdes.String(), Serdes.Long());
storePojoCount.toStream().to(TopicName.STREAM2.toString(), produced);
KafkaStreams streams = new KafkaStreams(builder.build(), properties);
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
streams.start();

产生了以下异常。

Exception in thread "cleant-id-StreamThread-2" org.apache.kafka.streams.errors.StreamsException: Deserialization exception handler is set to fail upon a deserialization error. If you would rather have the streaming pipeline continue after a deserialization error, please set the default.deserialization.exception.handler appropriately.
    at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:74)
    at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:91)
    at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)
    at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:546)
    at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:920)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:821)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744)
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

如何配置此 SpecificAvroSerde 以成功反序列化流？

【问题讨论】：

你为什么使用.groupBy((key, value) -> key)？由于您不更改密钥，因此最好使用groupByKey() - 这样可以避免昂贵的重新分区调用。我也不确定你自己的答案。从代码中我预计groupBy() 中会出现错误——因此，如果您更改为groupByKey()，它应该会得到解决。您对 default.value.serde 和 KTable 之间不匹配的总体观察是正确的——但是，对于 count() 操作，DSL 会自动将 serde 覆盖为 Long 类型作为已知的值类型。
@MatthiasJ.Sax 我同意 - 在这种情况下我应该使用 groupByKey。为了保持简洁，我简化了我的 SO 代码。

标签： java apache-kafka avro apache-kafka-streams confluent-platform

【解决方案1】：

问题在于物化对象没有适当的反序列化器 - Avro 正在尝试反序列化 KTable 值，因为 Avro 是默认值反序列化器。它无法这样做，因为 KTable 值实际上是 Longs。

使用正确的反序列化器创建物化对象将解决问题。

protected <K, V> Materialized<K, V, KeyValueStore<Bytes, byte[]>> persistentStore(StoreName storeName, Serde<K> keyType, Serde<V> valueType) {
    KeyValueBytesStoreSupplier storeSupplier = Stores.persistentKeyValueStore(storeName.toString());
    return Materialized.<K, V>as(storeSupplier).withKeySerde(keyType).withValueSerde(valueType);
}

任何商店供应商都可以在这里使用 - 这正是我需要的。

【讨论】：