Kafka Streams中Double类型反序列化异常问题求助

问题根源 #

报错Size of data received by Deserializer is not 8的核心原因是：

• Kafka的DoubleDeserializer要求输入必须是固定8字节的二进制Double值（Java中Double类型的原生二进制表示）。
• 你通过kafka-console-producer发送的是字符串形式的数字（如Test:12.34），这些内容被默认序列化为UTF-8字节数组，长度不等于8，因此触发反序列化失败。

即使你指定了value.serializer=org.apache.kafka.common.serialization.DoubleSerializer，console producer也无法将输入的字符串转换为二进制Double——因为DoubleSerializer仅接受Java Double对象作为输入，而控制台输入的是文本字符串，无法直接被该序列化器处理。

解决方案 #

方案一：自定义Serde处理字符串与Double的转换 #

如果需要继续使用kafka-console-producer发送文本格式的数据，可以自定义Serde，实现字符串到Double的解析与反向转换：

1. 实现自定义Serde

import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;
import java.util.Map;

public class DoubleFromStringSerde implements Serde<Double> {
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {}

    @Override
    public void close() {}

    @Override
    public Serializer<Double> serializer() {
        // 将Double转为字符串字节数组发送
        return (topic, data) -> data == null ? null : data.toString().getBytes();
    }

    @Override
    public Deserializer<Double> deserializer() {
        // 将字符串字节数组解析为Double
        return (topic, data) -> {
            if (data == null) return null;
            try {
                return Double.parseDouble(new String(data));
            } catch (NumberFormatException e) {
                throw new RuntimeException("Invalid Double string format", e);
            }
        };
    }
}

2. 修改Kafka Streams代码

将原代码中的Serdes.Double()替换为自定义Serde：

public static void main(String[] args) { 
    Properties properties = new Properties(); 
    properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");    
    properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    
    StreamsBuilder builder = new StreamsBuilder();

    // 使用自定义Serde消费和生产
    KStream<String, Double> inputTopic = builder.stream("a-topic",   
        Consumed.with(Serdes.String(), new DoubleFromStringSerde()));
    inputTopic.print(Printed.toSysOut());
                
    inputTopic.to("b-topic", Produced.with(Serdes.String(), new DoubleFromStringSerde()));
    
    Topology topology = builder.build();
    KafkaStreams kafkaStreams = new KafkaStreams(topology, properties);
    kafkaStreams.start();
    Runtime.getRuntime().addShutdownHook(new Thread(() -> kafkaStreams.close()));
} 

方案二：使用Java生产者发送原生二进制Double #

如果希望使用Kafka原生的DoubleSerializer/DoubleDeserializer，需要编写专门的Java生产者程序，直接发送Double类型的value：

Java生产者示例

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.DoubleSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class DoubleProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", DoubleSerializer.class.getName());

        try (Producer<String, Double> producer = new KafkaProducer<>(props)) {
            producer.send(new ProducerRecord<>("a-topic", "Test", 12.34));
            producer.send(new ProducerRecord<>("a-topic", "Test2", 123.454));
            producer.send(new ProducerRecord<>("a-topic", "Test3", 12345.321));
            System.out.println("Messages sent successfully");
        }
    }
}

这种情况下，你的原Kafka Streams代码无需修改——生产者发送的是8字节的二进制Double，完全符合DoubleDeserializer的要求。

内容的提问来源于stack exchange，提问作者IronMan