public boolean shouldDoubleWrite(long watermark) {
    // The writeLogstoreWatermarkGap is null, which means that the logstore writer is enabled
    // immediately once the job
    // is launched.
    if (writeLogstoreWatermarkGap == null) {
      LOG.info(
          "The logstore writer is enabled and the {} is null,"
              + " so double write immediately once the job is launched.",
          AUTO_EMIT_LOGSTORE_WATERMARK_GAP.key());
      return true;
    }
    long now = System.currentTimeMillis();
    boolean result = watermark >= now - writeLogstoreWatermarkGap.toMillis();
    if (result) {
      LOG.info(
          "The logstore writer is enabled and the {} is {}, the watermark has caught up, {}.",
          AUTO_EMIT_LOGSTORE_WATERMARK_GAP.key(),
          writeLogstoreWatermarkGap,
          watermark);
    } else {
      LOG.debug(
          "The logstore writer is enabled and the {} is {}, the watermark has not caught up, {}.",
          AUTO_EMIT_LOGSTORE_WATERMARK_GAP.key(),
          writeLogstoreWatermarkGap,
          watermark);
    }
    return result;
  }
Returns whether the automatic writing is enabled.

@param watermark the watermark of the operator
@return true: double write, false: single write.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/AutomaticWriteSpecification.java