KTable聚合后调用suppress存至中间主题遇类型转换异常求解决方案

问题背景 #

对聚合后的KTable<Windowed<String>, Object>执行suppress操作时，触发类型转换异常：

Caused by: java.lang.ClassCastException: class org.apache.kafka.streams.kstream.Windowed cannot be cast to class java.lang.String

场景是时间窗口较大（12/24小时）、数据量多，需要将结果存入其他主题，且必须保持初始的String serde配置直到执行suppress操作。相关代码片段如下：

split.branch((key, value) -> true, branchConsumer(s -> s
//.transform(TimestampTransformer::new)
.map((key, someObject) -> new KeyValue<>(someObject.getId(), someObject))
.groupByKey()
.windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(1), Duration.ofMinutes(1)))
.aggregate(NewObject::new,
        (key, value, aggregate) -> {
            // do the mapping between SomeObject and NewObject
            return aggregate;
        }
)
.suppress(Suppressed.<Windowed<String>>untilWindowCloses(Suppressed.BufferConfig.unbounded()
                .withLoggingEnabled(Map.of()))
        .withName("someName"))
// continue with what's left to do

问题原因 #

窗口聚合后的KTable键类型是Windowed<String>，但全局默认的键serde配置是String类型。Suppress操作在处理内部状态时会使用默认serde，导致尝试把Windowed<String>按String反序列化，触发类型转换异常。

解决办法 #

在调用suppress()之前，显式为KTable指定Windowed<String>类型的serde，覆盖全局默认配置即可，无需修改之前的String serde设置。具体步骤如下：

1. 先定义窗口实例，方便后续复用窗口大小参数
2. 在聚合后的KTable上调用withKeySerde()方法，传入对应类型的WindowedSerde

修改后的代码示例：

// 预先定义窗口实例，后续serde创建时复用窗口大小
TimeWindows timeWindows = TimeWindows.ofSizeAndGrace(Duration.ofMinutes(1), Duration.ofMinutes(1));

split.branch((key, value) -> true, branchConsumer(s -> s
//.transform(TimestampTransformer::new)
.map((key, someObject) -> new KeyValue<>(someObject.getId(), someObject))
.groupByKey()
.windowedBy(timeWindows)
.aggregate(NewObject::new,
        (key, value, aggregate) -> {
            // do the mapping between SomeObject and NewObject
            return aggregate;
        }
)
// 显式指定Windowed<String>类型的serde，覆盖全局默认配置
.withKeySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class, timeWindows.size()))
.suppress(Suppressed.<Windowed<String>>untilWindowCloses(Suppressed.BufferConfig.unbounded()
                .withLoggingEnabled(Map.of()))
        .withName("someName"))
// continue with what's left to do

如果后续需要将结果写入主题，也可以根据业务需求把Windowed<String>键转换成自定义格式（比如拼接窗口起始时间的字符串），但上述方法是最直接满足你“保持初始String serde直到suppress”要求的方案。

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