Hi Avi,
Just to verify your ITCase, I wrote the following dummy example and it
seems to be "working" (ie. I can see non null timestamps and timers firing).
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
env.setParallelism(1);
env
.addSource(new LongSource())
.keyBy(elmnt -> elmnt)
.process(new KeyedProcessFunction<Long, Long, Long>() {
@Override
public void processElement(Long value, Context ctx,
Collector<Long> out) throws Exception {
long timestamp = ctx.timestamp();
long timerTimestamp = timestamp + Time.seconds(10).toMilliseconds();
System.out.println(ctx.timestamp() + " " + timerTimestamp);
ctx.timerService().registerProcessingTimeTimer(timerTimestamp);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx,
Collector<Long> out) throws Exception {
System.out.println("TIMER: " + timestamp +" "+ ctx.timeDomain());
}
}).print();
env.execute();
The source is:
private static final class LongSource implements SourceFunction<Long> {
private volatile boolean running = true;
private long element = 0L;
@Override
public void run(SourceContext<java.lang.Long> ctx) throws Exception {
while (running) {
ctx.collect(element++ % 10);
Thread.sleep(10L);
}
}
@Override
public void cancel() {
}
}
Could you provide more details on how your usecase differs from the above
dummy example so that we can pin down the problem?
As a side-note, Ingestion time is essentially event time, with the only
difference that the timestamp assigner in the beginning gives each element
the timestamp System.currentTimeMillis. So in this case, maybe you could
also consider setting event time timers but keep in mind then your
Watermark emission interval.
In addition, if you want to simply check processing time processing of you
operator (not the whole pipeline), then you could make use of the
OneInputStreamTaskTestHarness or its keyed variant. This allows you to
provide your own processing time provider thus allow you to
deterministically
test processing time behaviour.
Cheers,
Kostas
On Sat, Mar 23, 2019 at 9:32 AM Avi Levi <avi.l...@bluevoyant.com> wrote:
> Any idea what should I do to overcome this?
>
> On Wed, Mar 20, 2019 at 7:17 PM Avi Levi <avi.l...@bluevoyant.com> wrote:
>
>> Hi Andrey,
>> I am testing a Filter operator that receives a key from the stream and
>> checks if it is a new one or not. if it is new it keeps it in state and
>> fire a timer all that is done using the ProcessFunction.
>> The testing is using some CollectSink as described here
>> <https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/testing.html#integration-testing>
>> and
>> the source is implementation of the SourceFunction that accepts a
>> collection of values and adds it to ctx.collect .
>> The ctx.timestamp() is null, BUT even if I set the timer to sometime in
>> the future ctx.timerService.registerProcessingTimeTimer(currenttimestamp +
>> x) the timer is fired immediately.
>>
>>
>> On Wed, Mar 20, 2019 at 10:39 AM Andrey Zagrebin <and...@ververica.com>
>> wrote:
>>
>>> Hi Avi,
>>>
>>> what is the structure of your unit test? do you create some source and
>>> then apply function or you test only ProcessFunction methods in isolation?
>>> does ctx.timestamp() return zero or which value?
>>>
>>> Best,
>>> Andrey
>>>
>>>
>>> On Tue, Mar 19, 2019 at 9:19 PM Avi Levi <avi.l...@bluevoyant.com>
>>> wrote:
>>>
>>>> Hi Andrey ,
>>>> I'm using IngestionTime
>>>> env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
>>>>
>>>> This is my timer in the processElement:
>>>> val nextTime: Long = ctx.timestamp() + daysInMilliseconds(14)
>>>> ctx.timerService.registerProcessingTimeTimer(nextTim)
>>>>
>>>> The problem is how do I use it in my unit tests ? since there is no
>>>> IngestionTime and timers are fired immediately so the timers actions (such
>>>> as state cleanup) are fired before time and causing the tests to fail .
>>>>
>>>>
>>>>
>>>>
>>>> On Tue, Mar 19, 2019 at 7:47 PM Andrey Zagrebin <and...@ververica.com>
>>>> wrote:
>>>>
>>>>> Hi Avi,
>>>>>
>>>>> do you use processing time timer
>>>>> (timerService().registerProcessingTimeTimer)?
>>>>> why do you need ingestion time? do you
>>>>> set TimeCharacteristic.IngestionTime?
>>>>>
>>>>> Best,
>>>>> Andrey
>>>>>
>>>>> On Tue, Mar 19, 2019 at 1:11 PM Avi Levi <avi.l...@bluevoyant.com>
>>>>> wrote:
>>>>>
>>>>>> Hi,
>>>>>> Our stream is not based on time sequence and we do not use time based
>>>>>> operations. we do want to clean the state after x days hence we fire
>>>>>> timer
>>>>>> event. My problem is that our unit test fires the event immediately
>>>>>> (there
>>>>>> is no ingestion time) how can I inject ingestion time ?
>>>>>>
>>>>>> Cheers
>>>>>> Avi
>>>>>>
>>>>>>
