如何停止程序中的閃存串流作業

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

根本的問題是，流媒體程序通常不是有限和無限地運行。

最好的辦法，至少就目前而言，是插入一個特殊的控制消息到您的流，它可以讓源正確地終止（簡單地停止而使讀循環讀取更多的數據）。這樣Flink就會告訴所有下游運營商，他們在消耗完所有數據之後可以停止。

或者，你可以在你的源拋出一個特殊的異常（例如一段時間後），這樣你可以（通過檢查錯誤原因）區分故障情況下的「正確」的終結。在源中拋出異常將導致程序失敗。

Answer 2

繼@TillRohrman

您可以將特殊的例外，方法和處理它在你的單元測試，如果你使用EmbeddedKafka實例，然後讀出了EmbeddedKafka主題，並斷言消費價值觀。

我發現https://github.com/asmaier/mini-kafka/blob/master/src/test/java/de/am/KafkaProducerIT.java是在這方面非常有用。

唯一的問題是，你將失去觸發異常的元素，但你可以隨時調整你的測試數據，以說明。