Kafka Streams:java.lang.IollegalArgumentException:VoidDeseri

Question

我正在处理我的第一个Kafka Streams样本：

package com.example;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
class DslExample {
public static void main(String[] args) {
// the builder is used to construct the topology
StreamsBuilder builder = new StreamsBuilder();
// read from the source topic, "users"
KStream<Void, String> stream = builder.stream("users");
// for each record that appears in the source topic,
// print the value
stream.foreach(
(key, value) -> {
System.out.println("(DSL) Hello, " + value);
});
// you can also print using the `print` operator
// stream.print(Printed.<String, String>toSysOut().withLabel("source"));
// set the required properties for running Kafka Streams
Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "dev1");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "10.0.0.24:29092");
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Void().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
// build the topology and start streaming
KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.start();
// close Kafka Streams when the JVM shuts down (e.g. SIGTERM)
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
}
}

当我试图运行它时，我得到了这个错误：

Caused by: java.lang.IllegalArgumentException: Data should be null for a VoidDeserializer.

这是来自"；用户"；主题：

值：

{
"registertime": 1517518703752,
"userid": "User_8",
"regionid": "Region_7",
"gender": "OTHER"
}

标题：

[
{
"key": "task.generation",
"stringValue": "0"
},
{
"key": "task.id",
"stringValue": "0"
},
{
"key": "current.iteration",
"stringValue": "86144"
}
]

密钥：

User_8

我应该怎么做才能避免这个问题？

Answer 1

如果密钥实际上有数据，则不应该使用Serdes.Void()或KStream<Void,

Answer 2

将Void更改为String将使流在示例中工作

KStream<String, String> stream = builder.stream("users");
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());