@Override
    public void stopTrackingAndReleaseJobPartitions(
            Collection<ResultPartitionID> partitionsToRelease) {
        LOG.debug("Releasing Job Partitions {}", partitionsToRelease);
        if (partitionsToRelease.isEmpty()) {
            return;
        }

        stopTrackingPartitions(partitionsToRelease);
        shuffleEnvironment.releasePartitionsLocally(partitionsToRelease);
    }
@Test
    void testStopTrackingAndReleaseJobPartitions() 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.stopTrackingAndReleaseJobPartitions(
                Collections.singleton(resultPartitionId1));

        assertThatFuture(shuffleReleaseFuture)
                .eventuallySucceeds()
                .satisfies(actual -> assertThat(actual).containsExactly(resultPartitionId1));
    }