我正在尝试为一个将数据写入kafka话题并从同一kafka话题读取数据的Flink流作业创建JUnit测试，使用FlinkKafkaProducer09和FlinkKafkaConsumer09。我在生产中传递了一个测试数据：
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中停止进程或运行特定时间？在测试中，您可以在单独的线程中启动作业执行，等待一些时间以允许进行数据处理，取消该线程（它将中断作业），然后进行断言。
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
if (deserializer.isEndOfStream(value)) {
// end of stream signaled
running = false;
break;
}
跟随@TillRohrman
如果您使用EmbeddedKafka实例，您可以结合特殊的异常方法并在单元测试中处理它，然后从EmbeddedKafka主题中读取并断言消费者值。
我发现https://github.com/asmaier/mini-kafka/blob/master/src/test/java/de/am/KafkaProducerIT.java 在这方面非常有用。
唯一的问题是您将失去触发异常的元素，但您始终可以调整测试数据以解决该问题。
我在@Mariusz W.给出的答案基础上使用了Awaitility来运行我的Flink应用程序，仅在本地测试期间直到满足某些条件，这应该缩短测试时间：
private static void runFlinkUntilConditionIsTrue() {
CompletableFuture<Void> handle = CompletableFuture.runAsync(() -> {
try {
env.execute();
} catch (Exception e) {
e.printStackTrace();
}
});
try {
Awaitility
.await()
.atMost(Duration.ofSeconds(30L)) // max duration for which to check the condition before giving up
.until(
// poll until some condition is met (or until max duration runs out)
// for me, this was getRecordsInLocalKinesisOutputTopic() == x
() -> someCondition()
);
} catch (ConditionTimeoutException e) {
System.out.println("Condition not met in time");
} finally {
try {
handle.get(0, TimeUnit.SECONDS);
} catch (TimeoutException | ExecutionException | InterruptedException e) {
// this will interrupt the job execution thread, cancel and close the job
handle.cancel(true);
}
}
}
Thread.sleep(10000)然后抛出一个已知消息的异常，3）try { env.execute(); } catch ...并检查异常消息，如果是已知消息，则吞下该异常；否则再次抛出该异常。 - winitzki