Thanks Gary,
I was only trying with a fixed set of events, so the Watermark was not
advancing, like you said.
Jayant Ameta
On Thu, Jan 11, 2018 at 3:36 PM, Gary Yao <g...@data-artisans.com> wrote:
> Hi Jayant,
>
> The difference is that the Watermarks from
> BoundedOutOfOrdernessTimestampExtractor are based on the greatest
> timestamp of
> all previous events. That is, if you do not receive new events, the
> Watermark
> will not advance. In contrast, your custom implementation of
> AssignerWithPeriodicWatermarks always advances the Watermark based on the
> wall
> clock.
>
> Maybe this will already help you to debug your application. If not, it
> would be
> great to see a minimal working example.
>
> Best,
> Gary
>
> On Wed, Jan 10, 2018 at 4:46 PM, Jayant Ameta <wittyam...@gmail.com>
> wrote:
>
>> Hi,
>> When using a BoundedOutOfOrdernessTimestampExtractor, the trigger is not
>> firing. However, the trigger fires when using custom timestamp extractor
>> with similar watermark.
>>
>> Sample code below:
>> 1.Assigner as anonymous class which works fine
>>
>> AssignerWithPeriodicWatermarks<Tuple2<Rule, T>> assigner = new
>> AssignerWithPeriodicWatermarks<Tuple2<Rule, T>>() {
>>
>> @Override
>> public long extractTimestamp(Tuple2<Rule, T> element, long
>> previousElementTimestamp) {
>> return System.currentTimeMillis();
>> }
>>
>> @Override
>> public final Watermark getCurrentWatermark() {
>> // this guarantees that the watermark never goes backwards.
>> return new Watermark(System.currentTimeMillis()-100);
>> }
>> };
>>
>>
>> 2.BoundedOutOfOrdernessTimestampExtractor assigner which doesn't work
>>
>> AssignerWithPeriodicWatermarks<Tuple2<Rule, T>> assigner = new
>> BoundedOutOfOrdernessTimestampExtractor<Tuple2<Rule,
>> T>>(Time.milliseconds(100)) {
>>
>> @Override
>> public long extractTimestamp(Tuple2<Rule, T> element) {
>> return System.currentTimeMillis();
>> }
>> };
>>
>>
>> Do you see any difference in the approaches?
>>
>> - Jayant
>>
>
>
