private Optional<MixedFormatSplit> getNextSplit(int subTaskId) {
    int currentParallelism = enumeratorContext.currentParallelism();
    if (totalParallelism != currentParallelism) {
      throw new FlinkRuntimeException(
          String.format(
              "Source parallelism has been changed, before parallelism is %s, now is %s",
              totalParallelism, currentParallelism));
    }
    if (subtaskSplitMap.containsKey(subTaskId)) {
      PriorityBlockingQueue<MixedFormatSplit> queue = subtaskSplitMap.get(subTaskId);

      MixedFormatSplit mixedFormatSplit = null;
      try {
        mixedFormatSplit = queue.poll(POLL_TIMEOUT, TimeUnit.MILLISECONDS);
      } catch (InterruptedException e) {
        LOG.warn("interruptedException", e);
      }
      if (mixedFormatSplit == null) {
        LOG.debug(
            "Subtask {}, couldn't retrieve mixed-format source split in the queue.", subTaskId);
        return Optional.empty();
      } else {
        LOG.info(
            "get next mixed-format split taskIndex {}, totalSplitNum {}, mixed-format split {}.",
            mixedFormatSplit.taskIndex(),
            totalSplitNum,
            mixedFormatSplit);
        return Optional.of(mixedFormatSplit);
      }
    } else {
      LOG.debug(
          "Subtask {}, it's an idle subtask due to the empty queue with this subtask.", subTaskId);
      return Optional.empty();
    }
  }
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