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);
序列化器实现:
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:
Caused by: 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(序列化器/反序列化器).可以克隆并试用: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 应用程序:
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屋!