@Override
    public List<ShuffleDescriptor> getClusterPartitionShuffleDescriptors(
            IntermediateDataSetID dataSetID) {
        final DataSetMetaInfo dataSetMetaInfo = this.dataSetMetaInfo.get(dataSetID);
        if (dataSetMetaInfo == null) {
            return Collections.emptyList();
        }

        return new ArrayList<>(dataSetMetaInfo.getShuffleDescriptors().values());
    }
@Test
    void testGetClusterPartitionShuffleDescriptors() {
        final ResourceManagerPartitionTrackerImpl tracker =
                new ResourceManagerPartitionTrackerImpl(new TestClusterPartitionReleaser());

        assertThat(tracker.listDataSets()).isEmpty();

        List<ResultPartitionID> resultPartitionIDS = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            resultPartitionIDS.add(
                    new ResultPartitionID(
                            new IntermediateResultPartitionID(DATA_SET_ID, i),
                            ExecutionAttemptID.randomId()));
        }

        for (ResultPartitionID resultPartitionID : resultPartitionIDS) {
            report(tracker, TASK_EXECUTOR_ID_1, DATA_SET_ID, 100, resultPartitionID);
        }

        final List<ShuffleDescriptor> shuffleDescriptors =
                tracker.getClusterPartitionShuffleDescriptors(DATA_SET_ID);
        assertThat(shuffleDescriptors).hasSize(100);
        assertThat(
                        shuffleDescriptors.stream()
                                .map(ShuffleDescriptor::getResultPartitionID)
                                .collect(Collectors.toList()))
                .containsExactlyElementsOf(resultPartitionIDS);

        reportEmpty(tracker, TASK_EXECUTOR_ID_1);
        reportEmpty(tracker, TASK_EXECUTOR_ID_2);
        assertThat(tracker.areAllMapsEmpty()).isTrue();
    }