是否指定了Kafka Streams拓扑的处理顺序？

Question

我想知道是否指定了流拓扑处理消息的顺序。

示例：

        // read input messages

        KStream<String, String> inputMessages = builder.stream("demo_input_topic_1");
        inputMessages = inputMessages.peek((k, v) -> System.out.println("TECHN. NEW MESSAGE: key: " + k + ", value: " + v));

        // check if message was already processed

        KTable<String, Long> alreadyProcessedMessages = inputMessages.groupByKey().count();
        KStream<String, String> newMessages =
                inputMessages.leftJoin(alreadyProcessedMessages, (streamValue, tableValue) -> getMessageValueOrNullIfKnownMessage(streamValue, tableValue));
        KStream<String, String> filteredNewMessages =
                newMessages.filter((key, val) -> val != null).peek((k, v) -> System.out.println("FUNC. NEW MESSAGE: key: " + k + ", value: " + v));

        // process the message

        filteredNewMessages.map((key, value) -> KeyValue.pair(key, "processed message: " + value))
                .peek((k, v) -> System.out.println("PROCESSED MESSAGE: key: " + k + ", value: " + v)).to("demo_output_topic_1");

使用GetMessagEvalueOrNullifKnownMessage（...）:

    private static String getMessageValueOrNullIfKnownMessage(String newMessageValue, Long messageCounter) {
        if (messageCounter > 1) {
            return null;
        }

        return newMessageValue;
    }

在测试中，该流工作。但我认为这不是保证。它只起作用，因为消息在加入之前首先由计数节点处理。

但是那个订单有保证吗？

就我在所有的文档中所看到的，对这个处理订单没有任何保证。因此,如果收到新消息,也可能发生以下情况：

null

谢谢你

Answer 1

没有保证。即使在当前的实现中，使用了子节点的list:https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/processorContexTimpl.java#L203-L206--但是，不能保证子节点按照DSL中指定的顺序追加到这个列表中（因为在这之间有一个转换层，它可以按照不同的顺序添加节点）。而且，实现可能在任何时间点发生更改。

我能想到的唯一可行的解决办法（这是相当昂贵的）是，在一个重复主题中发送流端数据：

KStream<String, String> newMessages =
   inputMessages.through(...) // note: as of 2.6.0 release, you could use `repartition()` instead of `through()`
                .leftJoin(alreadyProcessedMessages, ...);

这样，KTable将在连接执行之前更新，因为需要首先回读记录。但是，由于在回读记录时没有任何保证，因此在连接完成之前可能会对表进行多次更新，这可能会使您处于与以前类似的情况。（另外，通过另一个主题重新路由数据也有些昂贵。）

使用处理器API，您将拥有移动控制，因为您可以调用context.forward(...，to.child(...))。但是,对于这种情况,您还需要手动实现聚合和联接：

KStream routing = inputMessages.transform(...);
routing.groupByKey(...);
routing.leftJoin(...);

在本例中，您将在transform()之后获得要避免的重新分区主题：

KStream routing = inputMessages.transform(...);
routing.transform(...); // implement the aggregation
routing.transform(...); // implement the join