@Override
  public void handleSplitsChanges(SplitsChange<MixedFormatSplit> splitsChange) {
    if (!(splitsChange instanceof SplitsAddition)) {
      throw new UnsupportedOperationException(
          String.format("The SplitChange type of %s is not supported.", splitsChange.getClass()));
    }
    LOG.info("Handling a split change {}.", splitsChange);

    splitsChange
        .splits()
        .forEach(
            mixedFormatSplit -> {
              if (mixedFormatSplit instanceof SnapshotSplit
                  || mixedFormatSplit instanceof ChangelogSplit
                  || mixedFormatSplit instanceof MergeOnReadSplit) {
                splits.add(mixedFormatSplit);
              } else {
                throw new IllegalArgumentException(
                    String.format(
                        "As of now, The %s of SourceSplit type is unsupported, available source splits are %s, %s.",
                        mixedFormatSplit.getClass().getSimpleName(),
                        SnapshotSplit.class.getSimpleName(),
                        ChangelogSplit.class.getSimpleName()));
              }
            });
  }
A hybrid source split reader that could read {@link SnapshotSplit} and {@link ChangelogSplit}.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/reader/HybridSplitReader.java