@Override
  public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
    super.handleSourceEvent(subtaskId, sourceEvent);
    if (sourceEvent instanceof SplitRequestEvent) {
      SplitRequestEvent splitRequestEvent = (SplitRequestEvent) sourceEvent;
      Collection<String> finishedSplitIds = splitRequestEvent.finishedSplitIds();
      if (dimTable) {
        checkAndNotifyReader(finishedSplitIds);
      }
    } else if (sourceEvent instanceof ReaderStartedEvent) {
      if (!dimTable || temporalJoinSplits == null || !temporalJoinSplits.hasNotifiedReader()) {
        return;
      }
      // If tm failover, the reader may not be notified and watermark will not be retrieved in
      // reader.
      sourceEventBeforeFirstPlan = true;
      LOG.info("send InitializationFinishedEvent to reader again.");
      context.sendEventToSourceReader(subtaskId, InitializationFinishedEvent.INSTANCE);
    } else {
      throw new IllegalArgumentException(
          String.format(
              "Received unknown event from subtask %d: %s",
              subtaskId, sourceEvent.getClass().getCanonicalName()));
    }
  }
snapshotId for the last enumerated snapshot. next incremental enumeration should be based off
this as the starting position.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/enumerator/MixedFormatSourceEnumerator.java