ValueEventAssigner implements AssignerWithPunctuatedWatermarks<MRecord> { @Override public long extractTimestamp(MRecord event, long previousElementTimestamp) { long timestamp = event.getTimestamp(); currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp); return timestamp; } ValueEventAssigner(long delayMs); @Override long extractTimestamp(MRecord event, long previousElementTimestamp); @Override Watermark checkAndGetNextWatermark(MRecord event, long extractedTimestamp);  }
@Test public void extractTimestamp_works() { long extractTimestamp = assigner.extractTimestamp(UntypedValues.builder().timestamp(10L).build(), -1); assertEquals(10L, extractTimestamp); }