Kafka Streams 没有触发连接流的输出？答案

【问题标题】：Kafka Streams not triggering output for joined streams?Kafka Streams 没有触发连接流的输出？
【发布时间】：2018-10-07 20:32:01
【问题描述】：

我有来自 3 个 mysql 表、1 个主表和两个子表的原始流。我尝试加入三个原始流并转换为单个输出流。如果父流有任何更新，则它可以工作，但如果子流有任何变化，则不会触发输出。

    @StreamListener
    public Stream<Long, Output> handleStreams(@Input KStream<Long, Parent> parentStream,
    @Input KStream<Long, Child1> child1Stream,
    @Input KStream<Long, Child2> child2Stream) {

    KTable<Long, Parent> parentTable = convertParent(parentStream);
    KTable<Long, ArrayList<Child1>> child1Table = convertChild1(parentStream);
    KTable<Long, ArrayList<Child2>> child2Table = convertChild2(parentStream);

    parentTable
           .leftJoin(child1Table, (parent, child1List) -> new Output(k, v))
           .leftJoin(child2Table, (output, child2List) -> output.setChild2List(child2List))
           .toStream()
        }

父流上的任何新添加或更新都会被处理器拾取并将其与其他 KTable 连接并在输出流中返回。但是对 child1stream 或 child2stream 的任何添加或更新都不会触发输出流。

我认为将所有输入流设为 KTable，它们将始终存储更改，因为它们都具有相同的键，并且父表或子表上的任何更新都将被连接。但这并没有发生，任何人都可以建议我在这方面缺少什么吗？

我已经尝试过 KStream-KStream、Stream-KTable、KTable-KTable 连接，但在子更新的情况下它们都不起作用。

-谢谢

【问题讨论】：

标签： apache-kafka-streams spring-cloud-stream confluent-platform

【解决方案1】：

您能说明您在哪里拥有EnableBinding 和您要绑定的处理器接口吗？

这对我来说看起来不对：

@StreamListener
    public Stream<Long, Output> handleStreams(@Input KStream<Long, Parent> parentStream,
    @Input KStream<Long, Child1> child1Stream,
    @Input KStream<Long, Child2> child2Stream) {

您没有在输入上指定绑定。当您有多个输入时，您需要有这样的东西：

@StreamListener
        public Stream<Long, Output> handleStreams(@Input("input1") KStream<Long, Parent> parentStream,
        @Input("input2") KStream<Long, Child1> child1Stream,
        @Input("input3") KStream<Long, Child2> child2Stream) {

这些输入中的每一个都需要在处理器接口中定义。示例见此处：https://github.com/spring-cloud/spring-cloud-stream-samples/blob/master/kafka-streams-samples/kafka-streams-table-join/src/main/java/kafka/streams/table/join/KafkaStreamsTableJoin.java#L46

【讨论】：

对不起，这只是问题的伪表示。我有同样的代码，遵循同样的例子。有用。我的问题更多是关于一个概念。主流上的任何更新或插入都有效，但如果仅子流上的任何更改都不会触发输出对象。
如何从 mysql 接收原始流？子流从 kafka 主题中获取数据。您确定消息是从 mysql 到达那些 kafka 主题的吗？
是的，我正在使用带有 kafka 连接的 debezium。我在子流的分区聚合存储主题中看到新消息，但它们没有出现在输出中。
还是有点不清楚。上一条评论中的“输出”是什么意思？您是否看到流式传输到“child1Stream”和“child2stream”的消息？
是的，消息被流式传输到 child1stream 和 child2stream，并且还被流式传输到聚合存储。由于消息在聚合存储中（因为它们是 KTable），我希望它执行如上面代码所示的连接，并且应该生成一个输出对象流（它是主要、child1 和 child2 对象的组合）。

【解决方案2】：

注意您的子表是如何从与 parentTable 相同的流中创建的：

KTable<Long, ArrayList<Child1>> child1Table = convertChild1(parentStream);
KTable<Long, ArrayList<Child2>> child2Table = convertChild2(parentStream);

不确定 convertChild1 和 convertChild2 方法的作用，但不应该分别给它们 child1Stream 和 child2Stream 作为参数吗？

【讨论】：

对不起，你是对的。这是写这个问题的一个错字。也转换方法只是按键和聚合进行分组。子流可以有多个外键记录。