Flink Kafka：在没有收到消息的时间间隔后，优雅地关闭Flink消费来自Kafka源的消息

Question

我已将flinkkafkaconsumer作为源添加到我的streamexecutionenvironment中。我想在特定时间内没有收到新消息时关闭/阻止flink使用数据（类似于kafka polltime）。目前它正在无限期运行，并阻止执行移动到下一步（验证消息）。请建议是否有任何解决方法。

注意：我从反序列化中尝试了endofstream，但它无法工作，因为流实际上是不确定的。

提前谢谢。

Answer 1

如果这是为了测试，那么一种方法是创建自己的定制源代码，将FlinkKafkaConsumer包装起来。源代码的run（）方法将从线程调用Kafka源代码的run（）方法，传入一个封装真实收集器的收集器，并在收集任何内容时更新“上次收集的时间”。在源代码的run（）方法中，您将对此进行轮询，并在时间过长时调用Kakfa源代码的cancel（）方法，然后退出。

综上所述，通常在单元测试中，您希望使用模拟源代码来精确控制生成的内容、时间，而不是旋转Kafka系统。