@Override
    public void stopTrackingAndReleaseClusterPartitions(
            Collection<IntermediateDataSetID> dataSetsToRelease) {
        for (IntermediateDataSetID dataSetID : dataSetsToRelease) {
            final DataSetEntry dataSetEntry = clusterPartitions.remove(dataSetID);
            final Set<ResultPartitionID> partitionIds = dataSetEntry.getPartitionIds();
            shuffleEnvironment.releasePartitionsLocally(partitionIds);
        }
    }
@Test
    void stopTrackingAndReleaseClusterPartitions() throws Exception {
        final TestingShuffleEnvironment testingShuffleEnvironment = new TestingShuffleEnvironment();
        final CompletableFuture<Collection<ResultPartitionID>> shuffleReleaseFuture =
                new CompletableFuture<>();
        testingShuffleEnvironment.releasePartitionsLocallyFuture = shuffleReleaseFuture;

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

        final IntermediateDataSetID dataSetId1 = new IntermediateDataSetID();
        final IntermediateDataSetID dataSetId2 = new IntermediateDataSetID();

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

        partitionTracker.stopTrackingAndReleaseClusterPartitions(Collections.singleton(dataSetId1));
        assertThatFuture(shuffleReleaseFuture)
                .eventuallySucceeds()
                .satisfies(actual -> assertThat(actual).contains(resultPartitionId1));
    }