重新分区后Kafka流未使用serde

Question

我的Kafka Streams应用程序正在使用以下键值布局的kafka主题：String.class-

打印当前主题时，可以确认：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic flow-event-stream-file-service-test-instance --property print.key=true --property key.separator=" -- " --from-beginning
flow1 --  SUCCESS     #C:\Daten\file-service\in\crypto.p12

“flow1”是字符串键，--后面的部分是序列化值。

我的流程设置如下：

    KStream<String, HistoryEvent> eventStream = builder.stream(applicationTopicName, Consumed.with(Serdes.String(),
            historyEventSerde));


    eventStream.selectKey((key, value) -> new HistoryEventKey(key, value.getIdentifier()))
            .groupByKey()
            .reduce((e1, e2) -> e2,
                    Materialized.<HistoryEventKey, HistoryEvent, KeyValueStore<Bytes, byte[]>>as(streamByKeyStoreName)
                            .withKeySerde(new HistoryEventKeySerde()));

因此，据我所知，我告诉它使用String和历史事件序列来消费主题，因为这就是主题中的内容。然后，我'rekey'它使用一个组合的关键字，应该存储在本地使用提供的Serde为历史事件ey.class。据我所知，这将导致一个额外的主题被创建（可以在Kafka容器中的主题列表中看到）与新的关键字。这很好。

现在的问题是，即使在主题中只有一个文档的干净环境中，应用程序也无法启动：

org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_0, processor=KSTREAM-SOURCE-0000000000, topic=flow-event-stream-file-service-test-instance, partition=0, offset=0
Caused by: org.apache.kafka.streams.errors.StreamsException: A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: HistoryEventSerializer) is not compatible to the actual key or value type (key type: HistoryEventKey / value type: HistoryEvent). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.

从信息中很难判断问题的确切位置。它在我的基本主题中指出，但这是不可能的，因为那里的键不是类型HistoryEventKey。由于我在reduce中为HistoryEventKey提供了serde，因此它也不能与本地存储一起使用。

对我来说唯一有意义的是，它与导致重新排列和新主题的selectKey操作有关。但是，我无法确定如何为该操作提供serde。我不想将其设置为默认值，因为它不是默认的密钥serde。




Answer 1

我遇到了一条非常类似的错误消息，但我没有groupbys，而是joins。我在这里为谷歌搜索的下一个人发帖。

org.apache.kafka.streams.errors.StreamsException: ClassCastException while producing data to topic my-processor-KSTREAM-MAP-0000000023-repartition. A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: org.apache.kafka.common.serialization.StringSerializer) is not compatible to the actual key or value type (key type: java.lang.String / value type: com.mycorp.mySession). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters (for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with `Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).

很明显，和原来的问题一样，我不想更改默认的serdes。

所以在我的例子中，解决方案是在连接中传递一个连接实例，这将允许在SERDE中传递。请注意，错误消息指向重新分区映射- 这有点像转移视线，因为修复程序会转移到其他地方。

我如何修复它（一个加入的例子）

//...omitted ...

    KStream<String,MySession> mySessions = myStream
    .map((k,v) ->{
      MySession s = new MySession(v);
      k = s.makeKey();
      return new KeyValue<>(k, s);
    });
// ^ the mapping causes the repartition, you can not however specify a serde in there.


// but in the join right below, we can pass a JOINED instance and fix it.
    return enrichedSessions
      .leftJoin(
        myTable,
        (session, info) -> {
          session.infos = info;
          return session; },
        Joined.as("my_enriched_session")
              .keySerde(Serdes.String())
              .valueSerde(MySessionSerde())
      );

Answer 2

在对执行进行更多调试后，我能够发现新主题是在group pByKey步骤中创建的。您可以提供一个分组实例，它可以指定用于键和值的Serde：

    eventStream.selectKey((key, value) -> new HistoryEventKey(key, value.getIdentifier()))
            .groupByKey(Grouped.<HistoryEventKey, HistoryEvent>as(null)
                    .withKeySerde(new HistoryEventKeySerde())
                    .withValueSerde(new HistoryEventSerde())
            )
            .reduce((e1, e2) -> e2,
                    Materialized.<HistoryEventKey, HistoryEvent, KeyValueStore<Bytes, byte[]>>as(streamByKeyStoreName)
                            .withKeySerde(new HistoryEventKeySerde()));