@Override
    public void promoteJobPartitions(Collection<ResultPartitionID> partitionsToPromote) {
        LOG.debug("Promoting Job Partitions {}", partitionsToPromote);

        if (partitionsToPromote.isEmpty()) {
            return;
        }

        final Collection<PartitionTrackerEntry<JobID, TaskExecutorPartitionInfo>>
                partitionTrackerEntries = stopTrackingPartitions(partitionsToPromote);

        for (PartitionTrackerEntry<JobID, TaskExecutorPartitionInfo> partitionTrackerEntry :
                partitionTrackerEntries) {
            final TaskExecutorPartitionInfo dataSetMetaInfo = partitionTrackerEntry.getMetaInfo();
            final DataSetEntry dataSetEntry =
                    clusterPartitions.computeIfAbsent(
                            dataSetMetaInfo.getIntermediateDataSetId(),
                            ignored -> new DataSetEntry(dataSetMetaInfo.getNumberOfPartitions()));
            dataSetEntry.addPartition(partitionTrackerEntry.getMetaInfo().getShuffleDescriptor());
        }
    }
@Test
    void promoteJobPartitions() throws Exception {
        final TestingShuffleEnvironment testingShuffleEnvironment = new TestingShuffleEnvironment();
        final CompletableFuture<Collection<ResultPartitionID>> shuffleReleaseFuture =
                new CompletableFuture<>();
        testingShuffleEnvironment.releasePartitionsLocallyFuture = shuffleReleaseFuture;

        final JobID jobId = new JobID();
        final ResultPartitionID resultPartitionId1 = new ResultPartitionID();
        final ResultPartitionID resultPartitionId2 = new ResultPartitionID();

        final TaskExecutorPartitionTracker partitionTracker =
                new TaskExecutorPartitionTrackerImpl(testingShuffleEnvironment);
        partitionTracker.startTrackingPartition(
                jobId,
                new TaskExecutorPartitionInfo(
                        new TestingShuffleDescriptor(resultPartitionId1),
                        new IntermediateDataSetID(),
                        1));
        partitionTracker.startTrackingPartition(
                jobId,
                new TaskExecutorPartitionInfo(
                        new TestingShuffleDescriptor(resultPartitionId2),
                        new IntermediateDataSetID(),
                        1));
        partitionTracker.promoteJobPartitions(Collections.singleton(resultPartitionId1));

        partitionTracker.stopTrackingAndReleaseJobPartitionsFor(jobId);
        assertThatFuture(shuffleReleaseFuture)
                .eventuallySucceeds()
                .asList()
                .doesNotContain(resultPartitionId1);
    }