Hey, What's the recommended way to run integration...
# randomj
Joris Basiglio
12/01/2022, 12:34 AMHey, What's the recommended way to run integration test on a pipeline that rely on timer?
Should we somehow link test sinks and sources together so we can properly orchestrate events and watermarks in the desired order? Or is there an easier (maybe cleaner) approach to this problem?
a
Ari Huttunen
12/03/2022, 11:02 PMI really don’t know myself, but I asked OpenAI’s ChatGPT you verbatim question, and it wrote the following. Maybe someone knows if it got it correct?
One recommended way to run integration tests on a Flink pipeline that relies on timers is to use the TestStream source. The TestStream source allows you to simulate a stream of data and control the timestamps and watermarks of the data. This allows you to properly orchestrate events and watermarks in the desired order and test the behavior of your pipeline under different scenarios.
Here is an example of how you can use the TestStream source to test a Flink pipeline that uses timers:
(continued below)
Ari Huttunen
12/03/2022, 11:02 PM Copy code
// create a stream environment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// create a TestStream for the input data
TestStream<String> input = TestStream.create(
new TestData<String>()
.addElement("event1")
.addElement("event2")
.advanceWatermarkTo(10)
.addElement("event3")
.advanceWatermarkTo(20)
.addElement("event4")
);
// create the pipeline
DataStream<String> stream = env.addSource(input)
.map(...);
// create a TestSink for the output data
TestSink<String> output = new TestSink<>();
// connect the output sink to the pipeline
stream.addSink(output);
// execute the pipeline and retrieve the result from the output sink
env.execute();
List<String> result = output.getOutput();Ari Huttunen
12/03/2022, 11:03 PMIn this example, the TestStream source is used to simulate a stream of input data with controlled timestamps and watermarks. The pipeline is then executed using the execute method, and the output is retrieved from the TestSink sink. This allows you to test the behavior of your pipeline under different scenarios and verify that it produces the expected output.
Another approach you could consider is to use a testing framework such as JUnit or TestNG to write more structured and organized integration tests for your Flink pipeline. These frameworks provide various features and tools that can help you write, manage, and run your tests in a more efficient and effective way. You can use the TestStream source and the TestSink sink in combination with these testing frameworks to test your pipeline in a more organized and controlled manner.
s
sap1ens
12/04/2022, 1:19 AMUsing ChatGPT to answer questions about Flink? I feel like we hit bottom 🙂
😂 2
sap1ens
12/04/2022, 1:19 AMj
Joris Basiglio
12/05/2022, 10:51 PMThanks for the inputs, but my problem is that my pipeline is only partially keyed and there is a race condition between all the scenario messages as they are all being ingested at the same time. (when I would like to process some in a serial fashion.) Adding timer on top of that make the matter worst as sometime the last message get processed first and move the watermark which trigger all my timer instantly.
I resorted to use Thread.wait() to make sure my previous event were fully processed. Not great but I couldn't find anything better than that.
4 Views