@Override
    public ClusterPartitionReport createClusterPartitionReport() {
        List<ClusterPartitionReport.ClusterPartitionReportEntry> reportEntries =
                clusterPartitions.entrySet().stream()
                        .map(
                                entry ->
                                        new ClusterPartitionReport.ClusterPartitionReportEntry(
                                                entry.getKey(),
                                                entry.getValue().getTotalNumberOfPartitions(),
                                                entry.getValue().getShuffleDescriptors()))
                        .collect(Collectors.toList());

        return new ClusterPartitionReport(reportEntries);
    }
@Test
    void createClusterPartitionReport() {
        final TaskExecutorPartitionTrackerImpl partitionTracker =
                new TaskExecutorPartitionTrackerImpl(new NettyShuffleEnvironmentBuilder().build());

        assertThat(partitionTracker.createClusterPartitionReport().getEntries()).isEmpty();

        final IntermediateDataSetID dataSetId = new IntermediateDataSetID();
        final JobID jobId = new JobID();
        final ResultPartitionID clusterPartitionId = new ResultPartitionID();
        final ResultPartitionID jobPartitionId = new ResultPartitionID();
        final int numberOfPartitions = 1;

        partitionTracker.startTrackingPartition(
                jobId,
                new TaskExecutorPartitionInfo(
                        new TestingShuffleDescriptor(clusterPartitionId),
                        dataSetId,
                        numberOfPartitions));
        partitionTracker.startTrackingPartition(
                jobId,
                new TaskExecutorPartitionInfo(
                        new TestingShuffleDescriptor(jobPartitionId),
                        dataSetId,
                        numberOfPartitions + 1));

        partitionTracker.promoteJobPartitions(Collections.singleton(clusterPartitionId));

        final ClusterPartitionReport clusterPartitionReport =
                partitionTracker.createClusterPartitionReport();

        final ClusterPartitionReport.ClusterPartitionReportEntry reportEntry =
                Iterables.getOnlyElement(clusterPartitionReport.getEntries());
        assertThat(reportEntry.getDataSetId()).isEqualTo(dataSetId);
        assertThat(reportEntry.getNumTotalPartitions()).isEqualTo(numberOfPartitions);
        assertThat(reportEntry.getHostedPartitions()).contains(clusterPartitionId);
    }