We don't publish sources for test classes.
Have you considered that the sink will be serialized on job submission,
meaning that your myTestSink instance is not the one actually used by
the job? This typically means that have to store stuff in a static field
instead.
Alternatively, depending on the number of elements
org.apache.flink.streaming.api.datastream.DataStreamUtils#collect might
be worth a try.
On 15/05/2020 12:49, Manas Kale wrote:
Hi,
How do I test process functions? I tried by implementing a sink
function that stores myProcessFunction's output in a list. After
env.execute(), I use assertions.
If I set a breakpoint in the myTestSink's invoke() method, I see that
that method is being called correctly. However, after env.execute()
returns, all data in sink functions is wiped clean.
TestSink myTestSink = new myTestSink();
testStream.process(new myProcessFunction()).addSink(myTestSink);
env.execute("test");
assertEquals(expectedOutput, myTestSink.actual);
What am I doing wrong?
Also, I see that a ProcessFunctionTestHarnesses has been added in
1.10. I wasn't able to download its sources to understand how I could
use that. Have the sources not been added to maven or is it a problem
at my end?
Regards,
Manas
