Re: Watermark not worked as expected in TimeCharacteristic.EvenTime setting.

2018-03-03 Thread sundy
Thanks a lot, use env.setParallelism(1) before the source  define works (I set 
it before the env.execute, so it did not take effect).
 
> On 3 Mar 2018, at 16:02, sundy <543950...@qq.com> wrote:
> 
> Hi Hequn Cheng,
> 
> Finally I got the problem and find the way to define the correct WaterMark by 
> your advice, thank you very much.
> 
> The problem is that I set the watermark to the 
> 
> waterMark =  maxEventTime - lag 
> 
> And the timeWindow is 10Seconds,  But I generated the test records too 
> quickly so the 1 records are all in the window duration(my bad). 
> So flink are waiting for new more numbers to close the window.  
> 
> Another one question is why I set   'env.setParallelism(1)’ and run the code 
> in IDEA(mini Flink cluster) , but the getWatermark is called in 4 different 
> threads?
> Which time is  the getWaterMark function called? After the keyBy operation or 
> after the source operation?
> 
> 
> 
>> On 3 Mar 2018, at 15:28, Hequn Cheng > > wrote:
>> 
>> Hi sundy,
>> 
>> The default parallelism is 4. It seems that your configuration does not take 
>> effect. You can try 'env.setParallelism(1)' to set the job parallelism.
>> For watermark, you can refer to [1] and [2].
>> 
>> PS: Others can not see your reply because you only reply to me. Try reply to 
>> all so that others can help you too :-) 
>> 
>> [1] 
>> https://ci.apache.org/projects/flink/flink-docs-master/dev/event_time.html#watermarks-in-parallel-streams
>>  
>> 
>> [2] 
>> https://ci.apache.org/projects/flink/flink-docs-master/dev/event_timestamps_watermarks.html#timestamps-per-kafka-partition
>>  
>> 
>> 
>> On Sat, Mar 3, 2018 at 3:03 PM, sundy <543950...@qq.com 
>> > wrote:
>> 
>> Hi Hequn Cheng,
>> 
>> Thanks for you advice, I think I found the problem. But I didn't know why.
>> 
>> Firstly, let me introduce my operations,  I run the code in IDEA with 
>> MinFlinkCluster,  and then I set the env parallelism to 1, and the kafka 
>> topic ‘stream_test' has  one partition. Then send 1 records with current 
>> timestamp  the kafka , format is  ‘4281_783_1520059217832’ , the last field 
>> is the evenTime.
>> 
>> I add this debug code to print the thread-id.
>> 
>> 
>> 
>> 
>> The result is it will print in 4 threads int each period duration(3 
>> seconds).Such as
>> 
>> 50:water mark:0
>> 52:water mark:0
>> 51:water mark:1520056427871
>> 49:water mark:0
>> 
>> So this results to the watermark 0. But why it happened in 1 parallelism? 
>> Maybe it was caused by the keyBy operation?  I am new to Flink,  I hope to 
>> know how to set the watermark in the right way.
>> 
>> 
>>   
>>  
>> 
>>> On 3 Mar 2018, at 13:52, Hequn Cheng >> > wrote:
>>> 
>>> Hi sundy,
>>> 
>>> 1. Some partition of your input kafka don't have data. Since window 
>>> watermark is the min value of all it's inputs, if there are no data from 
>>> one of it's inputs, window will never be triggered. You can set parallelism 
>>> of your job to 1 to avoid this problem(PS: Maybe this bug is fixed now, but 
>>> worth a try).
>>> 2. Only one record in the input. In this case, window can not be triggered 
>>> either. You might think of it like the time has be stopped. To trigger the 
>>> widow, you should read more data with watermark bigger than the window end.
>>> 
>>> Hope it helps you.
>>> Best, Hequn
>>> 
>>> 2018-03-03 13:06 GMT+08:00 sundy <543950...@qq.com 
>>> >:
>>> Hi, thanks for your reply. 
>>> 
>>> I have searched it in stackoverflow, and there is someone who has the some 
>>> problem.
>>> 
>>> https://stackoverflow.com/questions/40993753/flink-streaming-program-runs-correctly-with-processing-time-but-will-not-produc
>>>  
>>> 
>>> 
>>> 
>>> From your advice, I tried the code. 
>>> 
>>>  env.getConfig().setAutoWatermarkInterval(3 * 1000);
>>> 
>>> And it calls the getCurrentWaterMark function each 3 seconds,  but still no 
>>> result come out.
>>> From the outputs   ('water mark1520049229163'), I could see that the add 
>>> method is called. But the no result from the sink function.
>>> 
>>> 
>>> 
>>> 
 On 3 Mar 2018, at 12:47, Xingcan Cui > wrote:
 
 Hi,
 
 for periodically generated watermarks, you should use 
 `ExecutionConfig.setAutoWatermarkInterval()` to set an interval.
 
 Hope that helps.
 
 Best,
 Xingcan
 
> On 3 Mar 2018, at 12:08 PM, sundy <543950...@qq.com 
> > wrote:
> 
> 
> 
> 

Re: Watermark not worked as expected in TimeCharacteristic.EvenTime setting.

2018-03-03 Thread Hequn Cheng
Hi sundy,

It is strange that your configuration does not take effect. Do you set
parallelism somewhere else? Maybe, you can refer to the kafka test case[1].
In this test case, line 229 set parallelism to 1 and works fine.
Hope it helps you.

[1]
https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java

On Sat, Mar 3, 2018 at 4:02 PM, sundy <543950...@qq.com> wrote:

> Hi Hequn Cheng,
>
> Finally I got the problem and find the way to define the correct WaterMark
> by your advice, thank you very much.
>
> The problem is that I set the watermark to the
>
> waterMark =  maxEventTime - lag
>
> And the timeWindow is 10Seconds,  But I generated the test records too
> quickly so the 1 records are all in the window duration(my bad).
> So flink are waiting for new more numbers to close the window.
>
> Another one question is why I set   'env.setParallelism(1)’ and run the
> code in IDEA(mini Flink cluster) , but the getWatermark is called in 4
> different threads?
> Which time is  the getWaterMark function called? After the keyBy operation
> or after the source operation?
>
>
>
> On 3 Mar 2018, at 15:28, Hequn Cheng  wrote:
>
> Hi sundy,
>
> The default parallelism is 4. It seems that your configuration does not
> take effect. You can try 'env.setParallelism(1)' to set the job
> parallelism.
> For watermark, you can refer to [1] and [2].
>
> PS: Others can not see your reply because you only reply to me. Try reply
> to all so that others can help you too :-)
>
> [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/
> event_time.html#watermarks-in-parallel-streams
> [2] https://ci.apache.org/projects/flink/flink-docs-master/dev/
> event_timestamps_watermarks.html#timestamps-per-kafka-partition
>
> On Sat, Mar 3, 2018 at 3:03 PM, sundy <543950...@qq.com> wrote:
>
>>
>> Hi Hequn Cheng,
>>
>> Thanks for you advice, I think I found the problem. But I didn't know why.
>>
>> Firstly, let me introduce my operations,  I run the code in IDEA with
>> MinFlinkCluster,  and then I set the env parallelism to 1, and the kafka
>> topic ‘stream_test' has  one partition. Then send 1 records with
>> current timestamp  the kafka , format is  ‘4281_783_1520059217832’ , the
>> last field is the evenTime.
>>
>> I add this debug code to print the thread-id.
>>
>> 
>>
>>
>> The result is it will print in 4 threads int each period duration(3
>> seconds).Such as
>>
>> 50:water mark:0
>> 52:water mark:0
>> 51:water mark:1520056427871
>> 49:water mark:0
>>
>> So this results to the watermark 0. But why it happened in 1 parallelism?
>> Maybe it was caused by the keyBy operation?  I am new to Flink,  I hope
>> to know how to set the watermark in the right way.
>>
>>
>>
>>
>>
>> On 3 Mar 2018, at 13:52, Hequn Cheng  wrote:
>>
>> Hi sundy,
>>
>> 1. Some partition of your input kafka don't have data. Since window
>> watermark is the min value of all it's inputs, if there are no data from
>> one of it's inputs, window will never be triggered. You can set
>> parallelism of your job to 1 to avoid this problem(PS: Maybe this bug is
>> fixed now, but worth a try).
>> 2. Only one record in the input. In this case, window can not be
>> triggered either. You might think of it like the time has be stopped. To
>> trigger the widow, you should read more data with watermark bigger than the
>> window end.
>>
>> Hope it helps you.
>> Best, Hequn
>>
>> 2018-03-03 13:06 GMT+08:00 sundy <543950...@qq.com>:
>>
>>> Hi, thanks for your reply.
>>>
>>> I have searched it in stackoverflow, and there is someone who has the
>>> some problem.
>>>
>>> https://stackoverflow.com/questions/40993753/flink-streaming
>>> -program-runs-correctly-with-processing-time-but-will-not-produc
>>>
>>>
>>> From your advice, I tried the code.
>>>
>>>  env.getConfig().setAutoWatermarkInterval(3 * 1000);
>>>
>>> And it calls the getCurrentWaterMark function each 3 seconds,  but still
>>> no result come out.
>>> From the outputs   ('water mark1520049229163'), I could see that the add
>>> method is called. But the no result from the sink function.
>>>
>>>
>>>
>>>
>>> On 3 Mar 2018, at 12:47, Xingcan Cui  wrote:
>>>
>>> Hi,
>>>
>>> for periodically generated watermarks, you should use
>>> `ExecutionConfig.setAutoWatermarkInterval()` to set an interval.
>>>
>>> Hope that helps.
>>>
>>> Best,
>>> Xingcan
>>>
>>> On 3 Mar 2018, at 12:08 PM, sundy <543950...@qq.com> wrote:
>>>
>>>
>>>
>>> Hi, I got a problem in Flink  and need your help.
>>>
>>> I tried to use TimeCharacteristic.EvenTime, but the sink function never
>>> be executed.
>>>
>>> public class StreamingJob {
>>>   public static void main(String[] args) throws Exception {
>>> // set up the streaming execution environment
>>> final StreamExecutionEnvironment env = 
>>> StreamExecutionEnvironment.getExecutionEnvironment();
>>> ObjectMapper 

Re: Watermark not worked as expected in TimeCharacteristic.EvenTime setting.

2018-03-03 Thread sundy
Hi Hequn Cheng,

Finally I got the problem and find the way to define the correct WaterMark by 
your advice, thank you very much.

The problem is that I set the watermark to the 

waterMark =  maxEventTime - lag 

And the timeWindow is 10Seconds,  But I generated the test records too quickly 
so the 1 records are all in the window duration(my bad). 
So flink are waiting for new more numbers to close the window.  

Another one question is why I set   'env.setParallelism(1)’ and run the code in 
IDEA(mini Flink cluster) , but the getWatermark is called in 4 different 
threads?
Which time is  the getWaterMark function called? After the keyBy operation or 
after the source operation?



> On 3 Mar 2018, at 15:28, Hequn Cheng  wrote:
> 
> Hi sundy,
> 
> The default parallelism is 4. It seems that your configuration does not take 
> effect. You can try 'env.setParallelism(1)' to set the job parallelism.
> For watermark, you can refer to [1] and [2].
> 
> PS: Others can not see your reply because you only reply to me. Try reply to 
> all so that others can help you too :-) 
> 
> [1] 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/event_time.html#watermarks-in-parallel-streams
>  
> 
> [2] 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/event_timestamps_watermarks.html#timestamps-per-kafka-partition
>  
> 
> 
> On Sat, Mar 3, 2018 at 3:03 PM, sundy <543950...@qq.com 
> > wrote:
> 
> Hi Hequn Cheng,
> 
> Thanks for you advice, I think I found the problem. But I didn't know why.
> 
> Firstly, let me introduce my operations,  I run the code in IDEA with 
> MinFlinkCluster,  and then I set the env parallelism to 1, and the kafka 
> topic ‘stream_test' has  one partition. Then send 1 records with current 
> timestamp  the kafka , format is  ‘4281_783_1520059217832’ , the last field 
> is the evenTime.
> 
> I add this debug code to print the thread-id.
> 
> 
> 
> 
> The result is it will print in 4 threads int each period duration(3 
> seconds).Such as
> 
> 50:water mark:0
> 52:water mark:0
> 51:water mark:1520056427871
> 49:water mark:0
> 
> So this results to the watermark 0. But why it happened in 1 parallelism? 
> Maybe it was caused by the keyBy operation?  I am new to Flink,  I hope to 
> know how to set the watermark in the right way.
> 
> 
>   
>  
> 
>> On 3 Mar 2018, at 13:52, Hequn Cheng > > wrote:
>> 
>> Hi sundy,
>> 
>> 1. Some partition of your input kafka don't have data. Since window 
>> watermark is the min value of all it's inputs, if there are no data from one 
>> of it's inputs, window will never be triggered. You can set parallelism of 
>> your job to 1 to avoid this problem(PS: Maybe this bug is fixed now, but 
>> worth a try).
>> 2. Only one record in the input. In this case, window can not be triggered 
>> either. You might think of it like the time has be stopped. To trigger the 
>> widow, you should read more data with watermark bigger than the window end.
>> 
>> Hope it helps you.
>> Best, Hequn
>> 
>> 2018-03-03 13:06 GMT+08:00 sundy <543950...@qq.com 
>> >:
>> Hi, thanks for your reply. 
>> 
>> I have searched it in stackoverflow, and there is someone who has the some 
>> problem.
>> 
>> https://stackoverflow.com/questions/40993753/flink-streaming-program-runs-correctly-with-processing-time-but-will-not-produc
>>  
>> 
>> 
>> 
>> From your advice, I tried the code. 
>> 
>>  env.getConfig().setAutoWatermarkInterval(3 * 1000);
>> 
>> And it calls the getCurrentWaterMark function each 3 seconds,  but still no 
>> result come out.
>> From the outputs   ('water mark1520049229163'), I could see that the add 
>> method is called. But the no result from the sink function.
>> 
>> 
>> 
>> 
>>> On 3 Mar 2018, at 12:47, Xingcan Cui >> > wrote:
>>> 
>>> Hi,
>>> 
>>> for periodically generated watermarks, you should use 
>>> `ExecutionConfig.setAutoWatermarkInterval()` to set an interval.
>>> 
>>> Hope that helps.
>>> 
>>> Best,
>>> Xingcan
>>> 
 On 3 Mar 2018, at 12:08 PM, sundy <543950...@qq.com 
 > wrote:
 
 
 
 Hi, I got a problem in Flink  and need your help.
 
 I tried to use TimeCharacteristic.EvenTime, but the sink function never be 
 executed.  
 
 public class StreamingJob {
   public static void main(String[] args) throws Exception {
 // set up the streaming execution environment
 final StreamExecutionEnvironment env = 
 

Re: Watermark not worked as expected in TimeCharacteristic.EvenTime setting.

2018-03-02 Thread Hequn Cheng
Hi sundy,

1. Some partition of your input kafka don't have data. Since window
watermark is the min value of all it's inputs, if there are no data from
one of it's inputs, window will never be triggered. You can set parallelism
of your job to 1 to avoid this problem(PS: Maybe this bug is fixed now, but
worth a try).
2. Only one record in the input. In this case, window can not be triggered
either. You might think of it like the time has be stopped. To trigger the
widow, you should read more data with watermark bigger than the window end.

Hope it helps you.
Best, Hequn

2018-03-03 13:06 GMT+08:00 sundy <543950...@qq.com>:

> Hi, thanks for your reply.
>
> I have searched it in stackoverflow, and there is someone who has the some
> problem.
>
> https://stackoverflow.com/questions/40993753/flink-streaming-program-runs-
> correctly-with-processing-time-but-will-not-produc
>
>
> From your advice, I tried the code.
>
>  env.getConfig().setAutoWatermarkInterval(3 * 1000);
>
> And it calls the getCurrentWaterMark function each 3 seconds,  but still
> no result come out.
> From the outputs   ('water mark1520049229163'), I could see that the add
> method is called. But the no result from the sink function.
>
>
>
>
> On 3 Mar 2018, at 12:47, Xingcan Cui  wrote:
>
> Hi,
>
> for periodically generated watermarks, you should use 
> `ExecutionConfig.setAutoWatermarkInterval()`
> to set an interval.
>
> Hope that helps.
>
> Best,
> Xingcan
>
> On 3 Mar 2018, at 12:08 PM, sundy <543950...@qq.com> wrote:
>
>
>
> Hi, I got a problem in Flink  and need your help.
>
> I tried to use TimeCharacteristic.EvenTime, but the sink function never be
> executed.
>
> public class StreamingJob {
>   public static void main(String[] args) throws Exception {
> // set up the streaming execution environment
> final StreamExecutionEnvironment env = 
> StreamExecutionEnvironment.getExecutionEnvironment();
> ObjectMapper jsonMapper = new ObjectMapper();
>
> Properties properties = new Properties();
> //String brokers = "172.27.138.8:9092";
> String brokers = "localhost:9092";
> properties.setProperty("bootstrap.servers", brokers);
> properties.setProperty("group.id", "test_fink");
> String topic = "stream_test";
>
> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
> FlinkKafkaConsumer010 myConsumer =
> new FlinkKafkaConsumer010(topic, new 
> BitRate.BitRateDeserializtionSchema(), properties);
>
> DataStream stream = env.addSource(myConsumer)
> .assignTimestampsAndWatermarks(new CustomWatermarkEmitter());
> DataStream
> reduceItems =
> stream
> .keyBy(a -> a.gameId)
> .timeWindow(Time.seconds(10))
> .reduce((a, b) -> a.add(b));
>
> reduceItems.addSink(new FlinkKafkaProducer010<>(brokers, "stream_sink", 
> (tuple) -> {
>   try {
> tuple.end();
> System.out.println(tuple.rate + "\t" + tuple.user);
> return jsonMapper.writeValueAsBytes(tuple);
>   } catch (JsonProcessingException e) {
> e.printStackTrace();
> return "".getBytes();
>   }
> }));
>
> env.execute("Flink Streaming Java API Skeleton");
>   }
>
> }
>
>
>
> Here is the CustomWatermarkEmitter. I tried to increase the lag number,
>  but not worked.
>
> public class CustomWatermarkEmitter implements 
> AssignerWithPeriodicWatermarks {
>
>   private long currentMax = 0;
>   private long lag = 3600 * 1000; //not worked ,even though the lag is very 
> big
>
>   @Nullable
>   @Override
>   public Watermark getCurrentWatermark() {
> long atLeastTime = currentMax - lag;
> System.out.println("water mark" + atLeastTime);
> return new Watermark(atLeastTime < 0 ? 0 : atLeastTime);
>   }
>
>   @Override
>   public long extractTimestamp(BitRate bitRate, long l) {
> currentMax = Long.max(bitRate.eventTime, currentMax);
> return bitRate.eventTime;
>   }
> }
>
>
>
> Here is the entity BitRate, the logs are generated in time , sample log
> `4281_783_1520047769115`
>
>
> public BitRate(long eventTime, long gameId, long rate, long user) {
>   this.eventTime = eventTime;
>
>   this.gameId = gameId;
>   this.rate = rate;
>   this.user = user;
>   this.startTs = System.currentTimeMillis();
>   this.type = 0;
> }
>
> public void end() {
>   this.endTs = System.currentTimeMillis();
> }
>
> public BitRate add(BitRate b) {
>   System.out.println("Add:" + b.rate);
>   this.rate += b.rate;
>   this.user += b.user;
>   return this;
> }
>
>
>
>
>


Re: Watermark not worked as expected in TimeCharacteristic.EvenTime setting.

2018-03-02 Thread sundy
Hi, thanks for your reply. 

I have searched it in stackoverflow, and there is someone who has the some 
problem.

https://stackoverflow.com/questions/40993753/flink-streaming-program-runs-correctly-with-processing-time-but-will-not-produc
 



From your advice, I tried the code. 

 env.getConfig().setAutoWatermarkInterval(3 * 1000);

And it calls the getCurrentWaterMark function each 3 seconds,  but still no 
result come out.
From the outputs   ('water mark1520049229163'), I could see that the add method 
is called. But the no result from the sink function.




> On 3 Mar 2018, at 12:47, Xingcan Cui  wrote:
> 
> Hi,
> 
> for periodically generated watermarks, you should use 
> `ExecutionConfig.setAutoWatermarkInterval()` to set an interval.
> 
> Hope that helps.
> 
> Best,
> Xingcan
> 
>> On 3 Mar 2018, at 12:08 PM, sundy <543950...@qq.com 
>> > wrote:
>> 
>> 
>> 
>> Hi, I got a problem in Flink  and need your help.
>> 
>> I tried to use TimeCharacteristic.EvenTime, but the sink function never be 
>> executed.  
>> 
>> public class StreamingJob {
>>   public static void main(String[] args) throws Exception {
>> // set up the streaming execution environment
>> final StreamExecutionEnvironment env = 
>> StreamExecutionEnvironment.getExecutionEnvironment();
>> ObjectMapper jsonMapper = new ObjectMapper();
>> 
>> Properties properties = new Properties();
>> //String brokers = "172.27.138.8:9092";
>> String brokers = "localhost:9092";
>> properties.setProperty("bootstrap.servers", brokers);
>> properties.setProperty("group.id ", "test_fink");
>> String topic = "stream_test";
>> 
>> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
>> FlinkKafkaConsumer010 myConsumer =
>> new FlinkKafkaConsumer010(topic, new 
>> BitRate.BitRateDeserializtionSchema(), properties);
>> 
>> DataStream stream = env.addSource(myConsumer)
>> .assignTimestampsAndWatermarks(new CustomWatermarkEmitter());
>> DataStream
>> reduceItems =
>> stream
>> .keyBy(a -> a.gameId)
>> .timeWindow(Time.seconds(10))
>> .reduce((a, b) -> a.add(b));
>> 
>> reduceItems.addSink(new FlinkKafkaProducer010<>(brokers, "stream_sink", 
>> (tuple) -> {
>>   try {
>> tuple.end();
>> System.out.println(tuple.rate + "\t" + tuple.user);
>> return jsonMapper.writeValueAsBytes(tuple);
>>   } catch (JsonProcessingException e) {
>> e.printStackTrace();
>> return "".getBytes();
>>   }
>> }));
>> 
>> env.execute("Flink Streaming Java API Skeleton");
>>   }
>> 
>> }
>> 
>> 
>> Here is the CustomWatermarkEmitter. I tried to increase the lag number,  but 
>> not worked.
>> 
>> public class CustomWatermarkEmitter implements 
>> AssignerWithPeriodicWatermarks {
>> 
>>   private long currentMax = 0;
>>   private long lag = 3600 * 1000; //not worked ,even though the lag is very 
>> big
>> 
>>   @Nullable
>>   @Override
>>   public Watermark getCurrentWatermark() {
>> long atLeastTime = currentMax - lag;
>> System.out.println("water mark" + atLeastTime);
>> return new Watermark(atLeastTime < 0 ? 0 : atLeastTime);
>>   }
>> 
>>   @Override
>>   public long extractTimestamp(BitRate bitRate, long l) {
>> currentMax = Long.max(bitRate.eventTime, currentMax);
>> return bitRate.eventTime;
>>   }
>> }
>> 
>> 
>> Here is the entity BitRate, the logs are generated in time , sample log   
>> `4281_783_1520047769115`
>> 
>> 
>> public BitRate(long eventTime, long gameId, long rate, long user) {
>>   this.eventTime = eventTime;
>> 
>>   this.gameId = gameId;
>>   this.rate = rate;
>>   this.user = user;
>>   this.startTs = System.currentTimeMillis();
>>   this.type = 0;
>> }
>> 
>> public void end() {
>>   this.endTs = System.currentTimeMillis();
>> }
>> 
>> public BitRate add(BitRate b) {
>>   System.out.println("Add:" + b.rate);
>>   this.rate += b.rate;
>>   this.user += b.user;
>>   return this;
>> }
>> 
> 



Re: Watermark not worked as expected in TimeCharacteristic.EvenTime setting.

2018-03-02 Thread Xingcan Cui
Hi,

for periodically generated watermarks, you should use 
`ExecutionConfig.setAutoWatermarkInterval()` to set an interval.

Hope that helps.

Best,
Xingcan

> On 3 Mar 2018, at 12:08 PM, sundy <543950...@qq.com> wrote:
> 
> 
> 
> Hi, I got a problem in Flink  and need your help.
> 
> I tried to use TimeCharacteristic.EvenTime, but the sink function never be 
> executed.  
> 
> public class StreamingJob {
>   public static void main(String[] args) throws Exception {
> // set up the streaming execution environment
> final StreamExecutionEnvironment env = 
> StreamExecutionEnvironment.getExecutionEnvironment();
> ObjectMapper jsonMapper = new ObjectMapper();
> 
> Properties properties = new Properties();
> //String brokers = "172.27.138.8:9092";
> String brokers = "localhost:9092";
> properties.setProperty("bootstrap.servers", brokers);
> properties.setProperty("group.id", "test_fink");
> String topic = "stream_test";
> 
> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
> FlinkKafkaConsumer010 myConsumer =
> new FlinkKafkaConsumer010(topic, new 
> BitRate.BitRateDeserializtionSchema(), properties);
> 
> DataStream stream = env.addSource(myConsumer)
> .assignTimestampsAndWatermarks(new CustomWatermarkEmitter());
> DataStream
> reduceItems =
> stream
> .keyBy(a -> a.gameId)
> .timeWindow(Time.seconds(10))
> .reduce((a, b) -> a.add(b));
> 
> reduceItems.addSink(new FlinkKafkaProducer010<>(brokers, "stream_sink", 
> (tuple) -> {
>   try {
> tuple.end();
> System.out.println(tuple.rate + "\t" + tuple.user);
> return jsonMapper.writeValueAsBytes(tuple);
>   } catch (JsonProcessingException e) {
> e.printStackTrace();
> return "".getBytes();
>   }
> }));
> 
> env.execute("Flink Streaming Java API Skeleton");
>   }
> 
> }
> 
> 
> Here is the CustomWatermarkEmitter. I tried to increase the lag number,  but 
> not worked.
> 
> public class CustomWatermarkEmitter implements 
> AssignerWithPeriodicWatermarks {
> 
>   private long currentMax = 0;
>   private long lag = 3600 * 1000; //not worked ,even though the lag is very 
> big
> 
>   @Nullable
>   @Override
>   public Watermark getCurrentWatermark() {
> long atLeastTime = currentMax - lag;
> System.out.println("water mark" + atLeastTime);
> return new Watermark(atLeastTime < 0 ? 0 : atLeastTime);
>   }
> 
>   @Override
>   public long extractTimestamp(BitRate bitRate, long l) {
> currentMax = Long.max(bitRate.eventTime, currentMax);
> return bitRate.eventTime;
>   }
> }
> 
> 
> Here is the entity BitRate, the logs are generated in time , sample log   
> `4281_783_1520047769115`
> 
> 
> public BitRate(long eventTime, long gameId, long rate, long user) {
>   this.eventTime = eventTime;
> 
>   this.gameId = gameId;
>   this.rate = rate;
>   this.user = user;
>   this.startTs = System.currentTimeMillis();
>   this.type = 0;
> }
> 
> public void end() {
>   this.endTs = System.currentTimeMillis();
> }
> 
> public BitRate add(BitRate b) {
>   System.out.println("Add:" + b.rate);
>   this.rate += b.rate;
>   this.user += b.user;
>   return this;
> }
>