Testing process functions

[Manas Kale]

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

[Chesnay Schepler]

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

[Manas Kale]

I see, I had not considered the serialization; that was the issue. 

Thank you.

On Mon, May 18, 2020 at 12:29 PM Chesnay Schepler <[hidden email]> wrote:

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

[Alexander Fedulov]

Hi Manas,

I would recommend using TestHarnesses for testing. You could also use them prior to 1.10. Here is an example of setting the dependencies:

https://github.com/afedulov/fraud-detection-demo/blob/master/flink-job/build.gradle#L113

You can see some examples of tests for a demo application here:

https://github.com/afedulov/fraud-detection-demo/blob/master/flink-job/src/test/java/com/ververica/field/dynamicrules/RulesEvaluatorTest.java

Hope this helps.

Best regards,

On Mon, May 18, 2020 at 1:18 PM Manas Kale <[hidden email]> wrote:

I see, I had not considered the serialization; that was the issue. 

Thank you.

On Mon, May 18, 2020 at 12:29 PM Chesnay Schepler <[hidden email]> wrote:

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

[Manas Kale]

Thank you for the example, Alexander.

On Wed, May 20, 2020 at 6:48 PM Alexander Fedulov <[hidden email]> wrote:

Hi Manas,

I would recommend using TestHarnesses for testing. You could also use them prior to 1.10. Here is an example of setting the dependencies:

https://github.com/afedulov/fraud-detection-demo/blob/master/flink-job/build.gradle#L113

You can see some examples of tests for a demo application here:

https://github.com/afedulov/fraud-detection-demo/blob/master/flink-job/src/test/java/com/ververica/field/dynamicrules/RulesEvaluatorTest.java

Hope this helps.

Best regards,

--

Alexander Fedulov | Solutions Architect

[phone]

Follow us @VervericaData

--

Join Flink Forward - The Apache Flink Conference

Stream Processing | Event Driven | Real Time

--

Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany

--

Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji (Tony) Cheng

On Mon, May 18, 2020 at 1:18 PM Manas Kale <[hidden email]> wrote:

I see, I had not considered the serialization; that was the issue. 

Thank you.

On Mon, May 18, 2020 at 12:29 PM Chesnay Schepler <[hidden email]> wrote:

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