tkaitchuck

@Override 
public void onProcessingTime(long timestamp) throws Exception { 
  // register next timer 
  Watermark newWatermark = userFunction.getCurrentWatermark(); 
  if (newWatermark != null && newWatermark.getTimestamp() > currentWatermark) { 
     currentWatermark = newWatermark.getTimestamp(); 
     // emit watermark 
     output.emitWatermark(newWatermark); 
  } 
  long now = getProcessingTimeService().getCurrentProcessingTime(); 

tkaitchuck

        EventStreamWriter<EventType> writer = clientFactory.createEventWriter(stream, serializer, 
                EventWriterConfig.builder().automaticallyNoteTime(true).build()); 

tkaitchuck

  Transaction<EventType> txn = writer.beginTxn(); 
  //... write events to transaction. 
  txn.commit(txnTimestamp); 

tkaitchuck

        EventStreamWriter<EventType> writer = clientFactory.createEventWriter(stream, serializer, 
                EventWriterConfig.builder().build()); 

        //... write events ... 

        writer.noteTime(currentTime); 

tkaitchuck

  final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
  env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);