覆盖 KStreams 默认序列化程序 (ByteArraySerializer)

Question

我似乎无法将主题的序列化程序覆盖为Serdes.String()。我正在尝试一个从主题（流）读取并写入 KTable 的简单用例。到目前为止我所拥有的：

@Component
class Processor {
    @Autowired
    public void process(final StreamsBuilder builder) {
        final Serde<String> stringSerde = Serdes.String();
        builder.stream("input_topic", Consumed.with(stringSerde, stringSerde))
                .filter((key, value) -> value.contains("ACTION"))
                .toTable(Materialized.as("output_table_materialized"))
                .toStream().to("output_table", Produced.with(stringSerde, stringSerde)); // EDIT: added this last line

    }
}

我得到的例外是：

org.apache.kafka.streams.errors.StreamsException: A serializer (org.apache.kafka.common.serialization.ByteArraySerializer) is not compatible to the actual key type (key type: java.lang.String). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.

据我所知，它知道消息是String，但它使用的是默认解串器ByteArraySerializer。我在上面的代码中哪里出错了？

Answer 1

Consumed.with 将是一个De序列化器。

错误出现在 Serializer 或 toTable 调用上，您可以添加 Produced.with 或修改您的应用程序属性以在那里配置默认值