@Override
  public InputStatus pollNext(ReaderOutput<T> output) throws Exception {
    this.output = output;
    emitWatermarkIfNeeded();
    return super.pollNext(wrapOutput(output));
  }
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