@Override
  public void onDiscoveredSplits(Collection<MixedFormatSplit> splits) {
    splits.forEach(this::putSplitIntoQueue);
    // only complete pending future if new splits are discovered
    completeAvailableFuturesIfNeeded();
  }
Key is subtaskId, Value is the queue of unAssigned mixed-format splits.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/assigner/ShuffleSplitAssigner.java