@Override
  public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
    if (sourceEvent instanceof SplitRequestEvent) {
      SplitRequestEvent splitRequestEvent = (SplitRequestEvent) sourceEvent;
      LOG.info("Received request split event from subtask {}", subtaskId);
      assigner.onCompletedSplits(splitRequestEvent.finishedSplitIds());
      readersAwaitingSplit.put(subtaskId, String.valueOf(splitRequestEvent.requesterHostname()));
      assignSplits();
    } else if (sourceEvent instanceof ReaderStartedEvent) {
      LOG.info("Received ReaderStartEvent from subtask {}", subtaskId);
    } else {
      throw new IllegalArgumentException(
          String.format(
              "Received unknown event from subtask %d: %s",
              subtaskId, sourceEvent.getClass().getCanonicalName()));
    }
  }
The abstract mixed-format source enumerator.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/enumerator/AbstractMixedFormatEnumerator.java