private void emitWatermarkIfNeeded() {
    if (this.output == null || !maxWatermarkToBeEmitted) {
      return;
    }
    LOGGER.info("emit watermark");
    output.emitWatermark(new Watermark(Long.MAX_VALUE));
    maxWatermarkToBeEmitted = false;
  }
SourceEvents may be received before this#pollNext.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/reader/MixedFormatSourceReader.java