code: public FlinkKafkaConsumerBase<T> assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks<T> assigner) {
  		checkNotNull(assigner);
  
  		if (this.periodicWatermarkAssigner != null) {
  			throw new IllegalStateException("A periodic watermark emitter has already been set.");
  		}
  		try {
  			ClosureCleaner.clean(assigner, true);
  			this.punctuatedWatermarkAssigner = new SerializedValue<>(assigner);
  			return this;
  		} catch (Exception e) {
  			throw new IllegalArgumentException("The given assigner is not serializable", e);
  		}
  	}
docs: Specifies an {@link AssignerWithPunctuatedWatermarks} to emit watermarks in a punctuated manner.
  The watermark extractor will run per Kafka partition, watermarks will be merged across partitions
  in the same way as in the Flink runtime, when streams are merged.
  
  <p>When a subtask of a FlinkKafkaConsumer source reads multiple Kafka partitions,
  the streams from the partitions are unioned in a "first come first serve" fashion. Per-partition
  characteristics are usually lost that way. For example, if the timestamps are strictly ascending
  per Kafka partition, they will not be strictly ascending in the resulting Flink DataStream, if the
  parallel source subtask reads more that one partition.
  
  <p>Running timestamp extractors / watermark generators directly inside the Kafka source, per Kafka
  partition, allows users to let them exploit the per-partition characteristics.
  
  <p>Note: One can use either an {@link AssignerWithPunctuatedWatermarks} or an
  {@link AssignerWithPeriodicWatermarks}, not both at the same time.
  
  @param assigner The timestamp assigner / watermark generator to use.
  @return The consumer object, to allow function chaining.