flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Till Rohrmann <trohrm...@apache.org>
Subject Re: Unit / Integration Test Timer
Date Fri, 14 Sep 2018 15:41:39 GMT
Hi Ashish,

how do you make sure that all of your data is not consumed within a
fraction of the 2 seconds? For this it would be better to use event time
which allows you to control how time passes. If you want to test a specific
operator you could try out the One/TwoInputStreamOperatorTestHarness.


On Fri, Sep 14, 2018 at 3:36 PM ashish pok <ashishpok@yahoo.com> wrote:

> All,
> Hopefully a quick one. I feel like I have seen this answered before a few
> times before but can't find an appropriate example. I am trying to run few
> tests where registered timeouts are invoked (snippet below). Simple example
> as show in documentation for integration test (using flink-test-utils)
> seems to complete even though Timers are registered and have not been
> invoked.
> StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
>         env.setParallelism(1);
>         CollectSink.values.clear();
>         // create a stream of custom elements and apply transformations
>         env.fromCollection(t.getTestInputs())
> .process(new TupleProcessFn())
> .keyBy(FactTuple::getKey)
> .process(new NormalizeDataProcessFn(2))
> .addSink(getSink())
>         env.execute();
> I have a 2 second processing timer registered. If I put a breakpoint in
> first TupleProcessFn() after a few Tuples are collected I can see onTimer
> being invoked. So what is the trick here? I went as far as putting in a
> MapFunction after second process function that has a sleep to no avail.
> Thanks,
> Ashish

View raw message