Spring Cloud Kafka流：无法获取WindowedAggregateSessions使用自定义Serdes

Question

我对使用kafka streams和spring cloud stream相对较新，在使用窗口聚合功能时遇到了困难。

我想做的是

1. 获取我的初始UserInteractionEvents流，并按userProjectId（字符串）对它们进行分组
2. 创建这些事件的窗口会话，15分钟不活动
3. 将这些窗口会话聚合到自定义会话对象中
4. 然后将这些会话对象转换为另一个自定义UserSession对象

我的代码是：

    @EnableBinding(KafkaStreamsProcessor::class)
    inner class SessionProcessorApplication {

        @StreamListener("input")
        @SendTo("output")
        fun process(input: KStream<*, UserInteractionEvent>): KStream<*, UserSession> {
            return input
                .groupBy({ _, v -> v.userProjectId }, Serialized.with(Serdes.String(), UserInteractionEventSerde()))
                .windowedBy(SessionWindows.with(TimeUnit.MINUTES.toMillis(15)))
                .aggregate(
                        Initializer<Session>(::Session),
                        Aggregator<String, UserInteractionEvent, Session> { _, event, session ->  session.interactions + event.interaction; session  },
                        Merger<String, Session> { _, session1, session2 ->  Session.merge(session1, session2)},
                        Materialized.`as`<String, Session, SessionStore<Bytes, ByteArray>>("windowed-sessions")
                        .withKeySerde(Serdes.String()).withValueSerde(SessionSerde()))
                .toStream()
                .map { windowed, session ->
                    KeyValue(windowed.key(),
                            UserSession(windowed.key(),
                                    session.interactions,
                                    Instant.ofEpochSecond(windowed.window().start()),
                                    Instant.ofEpochSecond(windowed.window().end()))
                    )
                }
        }
    }

我似乎在聚合部分有问题。在尝试刷新窗口会话存储时看到类强制转换异常。我很困惑如何从这里继续。如果有人能指出我哪里出错了，或者一些处理使用带有自定义服务器的会话窗口的留档，我将不胜感激！

非常感谢！

完整堆栈跟踪如下：

我的配置：

spring.cloud.stream.kafka.streams.bindings:
  default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
  default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
  input:
    consumer:
      valueSerde: com.teckro.analytics.UserInteractionEventSerde
  output:
    producer:
      valueSerde: com.teckro.analytics.UserSessionSerde

spring.cloud.stream.bindings:
  input:
    destination: test-interaction
    consumer:
      headerMode: raw
  output:
    destination: test-session
    producer:
      headerMode: raw

Answer 1

我发现您的配置存在一些问题。

默认Serde的配置方式应更改如下：

spring.cloud.stream.kafka.streams.binder.configuration:
  default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
  default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.bindings:
  input:
    consumer:
      valueSerde: com.teckro.analytics.UserInteractionEventSerde
  output:
    producer:
      valueSerde: com.teckro.analytics.UserSessionSerde

您似乎正在对所有反序列化/序列化使用本机Serde。您想在配置中包含它。默认情况下，绑定器执行输入/输出序列化。

spring.cloud.stream.bindings:
  input:
    destination: test-interaction
    consumer:
      useNativeDecoding: true
  output:
    destination: test-session
    producer:
      useNativeEncoding: true

如果问题仍然存在，请在Github上创建一个简单的示例项目并与我们分享。我们会看看。