KafkaStreams Serde异常 [英] KafkaStreams serde exception
问题描述
我正在使用Kafka和流技术；我已经为KStream创建了自定义序列化器和反序列化器，将用于接收来自给定主题的消息.
i am playing with Kafka and streams technology; i have created a custom serializer and deserializer for the KStream that i will use to receive messages from a given topic.
现在，问题在于我正在以这种方式创建Serde:
Now, the problem is that i am creating a serde in this way:
JsonSerializer<EventMessage> serializer = new JsonSerializer<>();
JsonDeserializer<EventMessage> deserializer = new JsonDeserializer<>(EventMessage.class);
Serde<EventMessage> messageSerde = Serdes.serdeFrom(serializer, deserializer);
Serializer的实现:
Serializer implementation:
public class JsonSerializer<T> implements Serializer<T> {
private Gson gson = new Gson();
public void configure(Map<String, ?> map, boolean b) {
}
@Override
public byte[] serialize(String topic, T data) {
return gson.toJson(data).getBytes(Charset.forName("UTF-8"));
}
@Override
public void close() {
}
}
反序列化器的实现:
public class JsonDeserializer<T> implements Deserializer<T> {
private Gson gson = new Gson();
private Class<T> deserializedClass;
public JsonDeserializer() {
}
public JsonDeserializer(Class<T> deserializedClass) {
this.deserializedClass = deserializedClass;
}
@Override
@SuppressWarnings("unchecked")
public void configure(Map<String, ?> map, boolean b) {
if(deserializedClass == null) {
deserializedClass = (Class<T>) map.get("serializedClass");
}
}
@Override
public T deserialize(String topic, byte[] data) {
System.out.print(data);
if(data == null){
return null;
}
return gson.fromJson(new String(data),deserializedClass);
}
@Override
public void close() {
}
}
当我尝试执行代码时，出现以下错误:
When i try to execute the code, i receive the following error:
由以下原因引起:org.apache.kafka.common.KafkaException:无法实例化org.apache.kafka.common.serialization.Serdes $ WrapperSerde类是否具有公共的无参数构造函数?
Caused by: org.apache.kafka.common.KafkaException: Could not instantiate class org.apache.kafka.common.serialization.Serdes$WrapperSerde Does it have a public no-argument constructor?
此处完整转储: https://pastebin.com/WwpuXuxB
这是我尝试使用Serde的方式:
This is the way i am trying to use serde:
KStreamBuilder builder = new KStreamBuilder();
KStream<String, EventMessage> eventsStream = builder.stream(stringSerde, messageSerde, topic);
KStream<String, EventMessage> outStream = eventsStream
.mapValues(value -> EventMessage.build(value.type, value.timestamp));
outStream.to("output");
此外，我不太确定我是否正确设置了属性以全局设置串行器和解串器:
Also, i am not totally sure i am setting up correctly the properties to setup up serializer and deserializer globally:
streamsConfiguration.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
streamsConfiguration.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, messageSerde.getClass());
推荐答案
要完成Matthias的答案，我刚刚编写了一个简单的示例，说明如何在Kafka Stream应用程序中创建自定义Serde(Serializer/Deserializer).可以克隆并尝试使用它: https://github.com/Davidcorral94/Kafka -Streams-Custom-Seder
To complete the Matthias answer I've just coded a simple example of how to create a custom Serde (Serializer / Deserializer) within a Kafka Stream App. It's is available to clone and try in: https://github.com/Davidcorral94/Kafka-Streams-Custom-Seder
首先，我创建两个类，一个用于序列化器，另一个用于反序列化器.在这种情况下，我使用 Gson库来执行序列化/反序列化.
First I create two classes, one for the Serializer and another for the Deserializer. In this case I use Gson library to perform the serialization/deserialization.
public class PersonSerializer implements Closeable, AutoCloseable, Serializer<Person> {
private static final Charset CHARSET = Charset.forName("UTF-8");
static private Gson gson = new Gson();
@Override
public void configure(Map<String, ?> map, boolean b) {
}
@Override
public byte[] serialize(String s, Person person) {
// Transform the Person object to String
String line = gson.toJson(person);
// Return the bytes from the String 'line'
return line.getBytes(CHARSET);
}
@Override
public void close() {
}
}
反序列化器
public class PersonDeserializer implements Closeable, AutoCloseable, Deserializer<Person> {
private static final Charset CHARSET = Charset.forName("UTF-8");
static private Gson gson = new Gson();
@Override
public void configure(Map<String, ?> map, boolean b) {
}
@Override
public Person deserialize(String topic, byte[] bytes) {
try {
// Transform the bytes to String
String person = new String(bytes, CHARSET);
// Return the Person object created from the String 'person'
return gson.fromJson(person, Person.class);
} catch (Exception e) {
throw new IllegalArgumentException("Error reading bytes", e);
}
}
@Override
public void close() {
}
}
然后，我将它们都包装到Serde中，以便能够将其用于我的Kafka Stream应用程序中.
Then, I wrap both of them into a Serde to be able to use it into my Kafka Stream App.
public class PersonSerde implements Serde<Person> {
private PersonSerializer serializer = new PersonSerializer();
private PersonDeserializer deserializer = new PersonDeserializer();
@Override
public void configure(Map<String, ?> configs, boolean isKey) {
serializer.configure(configs, isKey);
deserializer.configure(configs, isKey);
}
@Override
public void close() {
serializer.close();
deserializer.close();
}
@Override
public Serializer<Person> serializer() {
return serializer;
}
@Override
public Deserializer<Person> deserializer() {
return deserializer;
}
}
最后，您可以在下一行中将此Serde类用于您的Kafka Stream App:
Finally, you are able to use this Serde class into your Kafka Stream App with the next line:
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, PersonSerde.class);
这实际上与当前可用的最新Kafka版本1.0.0一起使用！
This is actually working with the latest Kafka version available at this moment which is 1.0.0!
这篇关于KafkaStreams Serde异常的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
