API for delayed/scheduled interval input source for integration tests

[Yee-Ning Cheng]

Hi,

I've been trying to write an integration test for my Flink application that has managed state with TTL expiration.

However, I can't seem to find a good way to create a stream of elements that waits X amount of time before each element is sent in.  I'm using the simple API:

val stream = env.fromElements(
  ...

  ...
)

Is there a way to do this? E.g. I'd like to wait 5 seconds before sending in the next element to see if the managed state expiration is working.

Thanks,

Yee-Ning

[Yee-Ning Cheng]

I was able to use the AbstractStreamOperatorTestHarness to write more of a
unit test for windowing operators.  However, I'm still trying to figure out
a way to have a "delayed iterator".  I tried implementing an iterator that
Thread.sleeps for the interval and passed it to the stream as an input, but
that didn't seem to work, plus I was having issues with serialization if I
enabled checkpointing which seemed like a hassle.



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

[Hequn Cheng]

Hi Yee,

Yes, AbstractStreamOperatorTestHarness is a good way to test an operator. As for iterator, do you use an IT or an UT test? I think Thread.sleeps may works for an IT test. If you use an UT, you probably need to set time by yourself, similar to setProcessingTime in harness test.

Best, Hequn

On Sat, Sep 1, 2018 at 12:20 PM Yee-Ning Cheng <[hidden email]> wrote:

I was able to use the AbstractStreamOperatorTestHarness to write more of a
unit test for windowing operators.  However, I'm still trying to figure out
a way to have a "delayed iterator".  I tried implementing an iterator that
Thread.sleeps for the interval and passed it to the stream as an input, but
that didn't seem to work, plus I was having issues with serialization if I
enabled checkpointing which seemed like a hassle.



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

[Dominik Wosiński]

Hey,
Maybe it would be a good idea to create somekind of test source for DataStream that allows writing elements to stream directly. Similarly like it’s done for reactive libraries sources. This would make creating tests a lot easier for Flink.

Best Regards,

Dom.

 

Wysłane z aplikacji Poczta dla Windows 10

 

Od: [hidden email]
Wysłano: sobota, 1 września 2018 17:18
Do: [hidden email]
DW: [hidden email]
Temat: Re: API for delayed/scheduled interval input source for integrationtests

 

Hi Yee,

 

Yes, AbstractStreamOperatorTestHarness is a good way to test an operator. As for iterator, do you use an IT or an UT test? I think Thread.sleeps may works for an IT test. If you use an UT, you probably need to set time by yourself, similar to setProcessingTime in harness test.

 

Best, Hequn

 

On Sat, Sep 1, 2018 at 12:20 PM Yee-Ning Cheng <[hidden email]> wrote:

I was able to use the AbstractStreamOperatorTestHarness to write more of a
unit test for windowing operators.  However, I'm still trying to figure out
a way to have a "delayed iterator".  I tried implementing an iterator that
Thread.sleeps for the interval and passed it to the stream as an input, but
that didn't seem to work, plus I was having issues with serialization if I
enabled checkpointing which seemed like a hassle.



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

 

[Yee-Ning Cheng]

I haven't tried it yet, but I saw flink-spector which seems to actually do
what I want..

https://github.com/ottogroup/flink-spector/wiki/Defining-Input

Although having it as part of the normal API would be nice as that library
is already out of date (1.5.0)



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

[Nicos Maris]

flink-spector is not really active and I have tried it and it is not as simple as it seams to be.

On Sat, Sep 1, 2018 at 10:45 PM Yee-Ning Cheng <[hidden email]> wrote:

I haven't tried it yet, but I saw flink-spector which seems to actually do
what I want..

https://github.com/ottogroup/flink-spector/wiki/Defining-Input

Although having it as part of the normal API would be nice as that library
is already out of date (1.5.0)



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/