In case where one needs t to use kafka event time ( ingestion time ) for watermark generation and timestamp extraction is setting EventTimeCharactersitic as EventTime enough ?
Or is this explicit code required ? consumer.assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<KafkaRecord>() { @Nullable @Override public Watermark checkAndGetNextWatermark(KafkaRecord lastElement, long extractedTimestamp) { return new Watermark(extractedTimestamp); } @Override public long extractTimestamp(KafkaRecord element, long previousElementTimestamp) { return previousElementTimestamp; } });