@Override
    public void releasePartitionsLocally(Collection<ResultPartitionID> partitionIds) {
        ioExecutor.execute(
                () -> {
                    for (ResultPartitionID partitionId : partitionIds) {
                        resultPartitionManager.releasePartition(partitionId, null);
                    }
                });
    }
@Test
    void testSlowIODoesNotBlockRelease() throws Exception {
        BlockerSync sync = new BlockerSync();
        ResultPartitionManager blockingResultPartitionManager =
                new ResultPartitionManager() {
                    @Override
                    public void releasePartition(ResultPartitionID partitionId, Throwable cause) {
                        sync.blockNonInterruptible();
                        super.releasePartition(partitionId, cause);
                    }
                };

        NettyShuffleEnvironment shuffleEnvironment =
                new NettyShuffleEnvironmentBuilder()
                        .setResultPartitionManager(blockingResultPartitionManager)
                        .setIoExecutor(Executors.newFixedThreadPool(1))
                        .build();

        shuffleEnvironment.releasePartitionsLocally(Collections.singleton(new ResultPartitionID()));
        sync.awaitBlocker();
        sync.releaseBlocker();
    }