使用键控协同处理函数的 Flink 超时 FlinkKafkaConsumer 的读取顺序

Question

我使用Flink数据流API中的< code > keyedcorprocessfunction 类来实现一个超时用例。场景如下:我有一个输入kafka主题和一个输出kafka主题，一个服务从输入主题中读取并处理它(持续可变的时间)，然后在输出Kafka主题中发布响应。

现在要实现超时（必须使用Flink datastream API），我有一个从kafka输入主题读取的FlinkKafka消费者和另一个从kafka输出主题读取的FlinkKafka消费者。我正在连接两个流，并使用Process Element1我正在注册一个计时器并等待onTimer方法被触发（声明了超时），或者Process Element2被触发，因此我删除了计时器并且没有声明超时。

在对大规模事件进行测试时，我看到＜code＞NULLPTREXCEPTION＜/code＞并且我怀疑＜code＞processElement2＜/code＞是在＜code＞ProcessElement1＜/code＞之前触发的，在上述情况下，由于任何原因（知道服务处理元素所花费的时间可能需要几秒），从输出主题读取元素的场景是否会发生在从输入主题读取元素之前？在这种情况下，严格使用Flink DataStream API实现上述超时功能的最佳情况是什么，请提供任何提示？

谢谢你。

Answer 1

是的，不能保证Flink会在Process Element2之前调用Process Element1方法。这很可能，但不确定。或者换句话说，可能并不总是这样。

这与您在Flink训练中的LongRideAlerts练习中发现的情况完全相同--https://github.com/apache/flink-training/tree/master/long-ride-alerts--因此，您可以模仿那里使用的解决方案。逻辑是这样的：

• 无论哪个事件（对于给定密钥）首先到达，从任一流中存储它
• 如果输入事件（来自主题 1）首先到达，请设置计时器
• 如果输出事件（来自 topic2）在计时器触发之前到达第二个，请删除计时器
• 无论哪个事件（对于给定密钥）从任一流中秒到达，请清除状态（由两个流共享）
• 每当计时器触发时，使用从主题 1 保存的输入事件以键控状态创建报告并清除状态

每当缺少来自 topic1 的输入事件时，此解决方案都会出现泄漏状态。如果重要，请参阅训练练习附带的讨论，了解如何处理该问题（简短版本：使用状态TTL）。