KafkaAvroDeserializer的Kafka流“consumed.with（）”

Question

我需要从KafkaAvroDeserializer而不是标准的kafka反序列化器消耗的主题创建一个流。这是因为它将被发送到汇流JDBC接收器连接器（不支持标准序列化器/反序列化器）中使用的主题。在创建主题时，我对key和value都使用了KafkaAvroSerializer。

我的原始代码（在更改为密钥使用Kafka Avro序列化器之前）是：

final KStream<String, DocumentUpload> uploadStream = builder.stream(UPLOADS_TOPIC, Consumed.with(Serdes.String(), uploadSerde));

注意：上面的Serdes.String不会正确反序列化，因为密钥是使用KafKaavroSerializer序列化的。那么，也许有另一种形式的代码，它允许我构建一个流，而不需要设置关键字serdes（因此它默认为配置中的内容），我可以只设置值serde（uploadSerde）？

如果不是，有人能告诉我如何将“serdes.string（）”标准deserizlaizer更改为kafkaavrodeserializer吗？例如。

final KStream<String, DocumentUpload> uploadStream = builder.stream(UPLOADS_TOPIC, Consumed.with(<What can I insert here for the KafkaAvroDeserializer.String???>, uploadSerde));

在我的使用者中，我正在设置正确的默认反序列化器：

streamsConfiguration.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroDeserializer.class);
streamsConfiguration.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroDeserializer.class);

如果使用表单（并允许在我的消费者中指定的默认值，即KafkaAvro）：

final KStream<String, DocumentUpload> uploadStream = builder.stream(UPLOADS_TOPIC);

我得到以下信息：

2018-04-08 00:24:53,433] ERROR [fc-demo-client-StreamThread-1] stream-thread [fc-demo-client-StreamThread-1] Failed to process stream task 0_0 due to the following error:    (org.apache.kafka.streams.processor.internals.AssignedTasks)
java.lang.ClassCastException: [B cannot be cast to java.lang.String
at     org.apache.kafka.streams.kstream.internals.KStreamPeek$KStreamPeekProcessor.process(KStreamPeek.java:42)
at    org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:46)
at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208)
....

多谢了。

Answer 1

键和值的逻辑是相同的。因此，您可以以相同的方式处理这两种情况。

您的困惑是在配置中设置使用者反序列化器。请注意，这些配置被忽略（出于内部原因）。不能直接配置使用者的反序列化器。你总是需要用Serdes。因此，如果要为使用者设置默认的反序列化程序，则需要在配置中指定默认的Serde。

因此，我创建了一个围绕KafkaAvroSerializer和KafkaAvroDeserializer的包装器，该包装器实例化了它们，然后将包装器用于consumed.with中的键参数

完全正确。也可以在配置中将此Serde设置为默认值。

我本以为使用KafKaavroSerialize的字符串键从主题创建流是一种常见的用例

对此不确定。如果是纯字符串，我假设人们可能直接使用StringDeserializer，而不是将字符串包装为Avro（不确定）。还要注意，在处理Avro时，建议使用模式注册表。Confluent的模式注册表附带相应的Avro Serdes：https://github.com/confluentinc/schema-registry/（免责声明：我是Confluent的雇员。）