Testing process functions

classic Classic list List threaded Threaded
5 messages Options
Reply | Threaded
Open this post in threaded view
|

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
Reply | Threaded
Open this post in threaded view
|

Re: Testing process functions

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


Reply | Threaded
Open this post in threaded view
|

Re: Testing process functions

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


Reply | Threaded
Open this post in threaded view
|

Re: Testing process functions

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:

You can see some examples of tests for a demo application here:
Hope this helps.

Best regards,

--

Alexander Fedulov | Solutions Architect

+49 1514 6265796



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


Reply | Threaded
Open this post in threaded view
|

Re: Testing process functions

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:

You can see some examples of tests for a demo application here:
Hope this helps.

Best regards,

--

Alexander Fedulov | Solutions Architect

+49 1514 6265796



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