@Override
    public void stopTrackingAndReleaseAllClusterPartitions() {
        clusterPartitions.values().stream()
                .map(DataSetEntry::getPartitionIds)
                .forEach(shuffleEnvironment::releasePartitionsLocally);
        clusterPartitions.clear();
    }
@Test
    void stopTrackingAndReleaseAllClusterPartitions() 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 TaskExecutorPartitionTracker partitionTracker =
                new TaskExecutorPartitionTrackerImpl(testingShuffleEnvironment);
        partitionTracker.startTrackingPartition(
                new JobID(),
                new TaskExecutorPartitionInfo(
                        new TestingShuffleDescriptor(resultPartitionId1),
                        new IntermediateDataSetID(),
                        1));
        partitionTracker.startTrackingPartition(
                new JobID(),
                new TaskExecutorPartitionInfo(
                        new TestingShuffleDescriptor(resultPartitionId2),
                        new IntermediateDataSetID(),
                        1));
        partitionTracker.promoteJobPartitions(Collections.singleton(resultPartitionId1));

        partitionTracker.stopTrackingAndReleaseAllClusterPartitions();
        assertThatFuture(shuffleReleaseFuture)
                .eventuallySucceeds()
                .satisfies(actual -> assertThat(actual).contains(resultPartitionId1));
    }