Kafka Streams没有为加入的流触发输出？

Question

我有来自 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，它们将始终html" target="_blank">存储更改，因为它们都具有相同的键，并且父表或子表的任何更新都将被选取连接。但它没有发生，谁能建议我在这里错过了什么？

我已经尝试了KStream-KStream、Stream-KTable、KTable-KTable连接，没有一个能在子更新的情况下工作。

-谢谢

Answer 1

注意子表是如何从与父表相同的流中创建的:

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

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

Answer 2

您可以显示EnableB的和您要绑定到的处理器接口的位置吗？

这在我看来不太对劲：

@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