如何从程序中停止flink流作业

Question

我正在尝试为Flink流媒体作业创建JUnit测试，该作业将数据写入kafka主题，并分别使用FlinkKafkaProducer09和FlinkKafkaConsumer09从同一kafka主题读取数据。我正在通过生产中的测试数据：

DataStream<String> stream = env.fromElements("tom", "jerry", "bill");

以及检查来自消费者的数据是否与以下数据相同：

List<String> expected = Arrays.asList("tom", "jerry", "bill");
List<String> result =  resultSink.getResult();
assertEquals(expected, result);

使用TestListResultSink。

通过打印流，我能够看到来自消费者的数据。但无法获得Junit测试结果，因为即使消息完成，使用者仍将继续运行。所以它并没有来测试这个部件。

在Flink或FlinkKafkaConsumer09中是否有任何方法停止进程或运行特定时间？

Answer 1

在您的测试中，您可以在单独的线程中启动作业执行，等待一段时间以允许它进行数据处理，取消线程（它将中断作业）并进行关联。

CompletableFuture<Void> handle = CompletableFuture.runAsync(() -> {
    try {
        environment.execute(jobName);
    } catch (Exception e) {
        e.printStackTrace();
    }
});
try {
    handle.get(seconds, TimeUnit.SECONDS);
} catch (TimeoutException e) {
    handle.cancel(true); // this will interrupt the job execution thread, cancel and close the job
}

// Make assertions here

Answer 2

您不能在反序列化程序中使用isEndOfStream覆盖来停止从Kafka提取吗？如果我读得正确，flink/Kafka09Fetcher的run方法中有以下代码，这会中断事件循环

    if (deserializer.isEndOfStream(value)) {
                        // end of stream signaled
                        running = false;
                        break;
                    }

我的想法是使用Till Rohrmann关于控制消息的想法，并结合这个isEndOfStream方法来告诉Kafka消费者停止阅读。

有什么理由不起作用吗？或者是一些我忽略的角落案例？

https://github.com/apache/flink/blob/07de86559d64f375d4a2df46d320fc0f5791b562/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java#L146

Answer 3

潜在的问题是，流媒体节目通常不是有限的，而且运行无限期。

最好的方法，至少目前是这样，是在流中插入一个特殊的控制消息，让源正确终止（只需通过离开读取循环停止读取更多数据）。这样，Flink将告诉所有下游操作员，他们可以在消耗完所有数据后停止。

或者，您可以在源中抛出一个特殊的异常（例如，在一段时间后），这样您就可以通过检查错误原因来区分“正确”终止和失败情况。在源代码中抛出异常将使程序失败。