Kafka流错误：SerializationException：LongDeserializer接收到的数据大小不是8

Question

我正在尝试Kafka Streams。编写一个简单的应用程序，我在其中计算重复的消息。

消息：

2019-02-27-11:16:56 :: session:prod-111656 :: Msg => Hello World: 2491
2019-02-27-11:16:56 :: session:prod-111656 :: Msg => Hello World: 2492

等。

我正在尝试通过会话拆分此类消息：prod xxxx。用它作为钥匙。然后会话：prod xxxx Hello World:xxxx将其用作值。然后按键分组，查看在每个会话中复制了哪些消息。

这是代码：

KStream<String, String> textLines = builder.stream("RegularProducer");
KTable<String, Long> ktable = textLines.map(
    (String key, String value) -> {
        try {
            String[] parts = value.split("::");
            String sessionId = parts[1];
            String message = ((parts[2]).split("=>"))[1];
            message = sessionId+":"+message;
            return new KeyValue<String,String>(sessionId.trim().toLowerCase(), message.trim().toLowerCase());
        } catch (Exception e) {
            return new KeyValue<String,String>("Invalid-Message".trim().toLowerCase(), "Invalid Message".trim().toLowerCase());
        }
    })
    .groupBy((key,value) -> value)
    .count().filter(
            (String key, Long value) -> {
                return value > 1;
            }
    );

ktable.toStream().to("RegularProducerDuplicates", 
Produced.with(Serdes.String(), Serdes.Long()));
Topology topology = builder.build();
topology.describe();
KafkaStreams streams = new KafkaStreams(topology, props);
streams.start();

KTable 主题 常规生产者生成重复。但是，当我使用控制台使用者查看它时，它会崩溃并显示错误。然后，我在控制台使用者上使用 --跳过错误消息标志。现在我看到成千上万的这样的线条

session:prod-111656 : hello world: 994  [2019-02-28 16:25:18,081] ERROR Error processing message, skipping this message:  (kafka.tools.ConsoleConsumer$)
org.apache.kafka.common.errors.SerializationException: Size of data received by LongDeserializer is not 8

谁能帮帮我这里出了什么问题？

Answer 1

我也有同样的问题，发现如果我将filter移到toStream之后，就不会产生空值（墓碑）。

Answer 2

用于值的反序列化程序可能不用于字符串，而将用于长整型。在cli中创建使用者时，请务必指定它。表示“出”

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
    --topic name \
    --from-beginning \
    --formatter kafka.tools.DefaultMessageFormatter \
    --property print.key=true \
    --property print.value=true \
    --skip-message-on-error \
    --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
    --property value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

在创建使用者时，请在此处检查最后 2 行，注意您的类型（键，值） 在我的情况下，两者都是字符串，如果值是长类型，请使用最后一行作为： --属性值.反序列化器=org.apache.kafka.common.序列化.长序列化器

Answer 3

您的Kafka Streams应用程序可以正常工作。

该错误位于kafka-consolt-消费者（kafka.tools.ConsoleConsumer是实现脚本逻辑的类）中。

在反序列化期间，它无法正确处理null。当它获取null作为消息的值或键时，它会设置默认值（表示nullString的字节数组）。如果检查源代码，您可以找到以下函数

def write(deserializer: Option[Deserializer[_]], sourceBytes: Array[Byte]) {
  val nonNullBytes = Option(sourceBytes).getOrElse("null".getBytes(StandardCharsets.UTF_8))
  val convertedBytes = deserializer.map(_.deserialize(null, nonNullBytes).toString.
    getBytes(StandardCharsets.UTF_8)).getOrElse(nonNullBytes)
  output.write(convertedBytes)
}

您如何查看它何时获取null的source ceBytes（source ceBytes==null）以进行反序列化，它为此设置了默认值：

val nonNullBytes=Option（source ceBytes）. getOrElse（"null". getBytes（StandardCharsets.UTF_8））

在您的示例中，它是“null”.getBytes（StandardCharsets.UTF_8）。然后，尝试使用org.apache.kafka.common.serialization进行反序列化。LongDeserializer（值反序列化器）LongDeserializer在开始时检查字节数组的大小。现在是4（null的字节表示形式），并引发异常。

例如，如果您使用StringDeserializer，它不会正确反序列化它，但至少它不会引发异常，因为它不检查字节数组的长度。

长话短说：ConsoleConsumer的格式化程序，即负责打印，对于漂亮的打印设置了一些默认值，这些默认值无法由某些反序列化程序（长序列化器，整数序列化程序）处理

关于，为什么应用程序为某些键生成null值：

KTable:filter与KStream:：filter具有不同的语义。根据KTable的javadoc：

对于每个被删除的记录（即不满足给定谓词），将转发一个逻辑删除记录。

对于您的< code >过滤器，当< code >计数