public ChannelStateWriteRequestExecutor getOrCreateExecutor(
            JobVertexID jobVertexID,
            int subtaskIndex,
            SupplierWithException<CheckpointStorageWorkerView, ? extends IOException>
                    checkpointStorageWorkerViewSupplier,
            int maxSubtasksPerChannelStateFile) {
        return getOrCreateExecutor(
                jobVertexID,
                subtaskIndex,
                checkpointStorageWorkerViewSupplier,
                maxSubtasksPerChannelStateFile,
                true);
    }
@Test
    void testSomeSubtasksCloseDuringOtherSubtasksStarting() throws Exception {
        JobID jobID = new JobID();
        JobVertexID jobVertexID = new JobVertexID();
        int numberOfSubtask = 100_000;
        int maxSubtasksPerChannelStateFile = 10;

        ChannelStateWriteRequestExecutorFactory executorFactory =
                new ChannelStateWriteRequestExecutorFactory(jobID);

        BlockingQueue<ChannelStateWriteRequestExecutor> queue = new LinkedBlockingQueue<>(100);

        CompletableFuture<Void> createFuture = new CompletableFuture<>();
        new Thread(
                        () -> {
                            try {
                                for (int i = 0; i < numberOfSubtask; i++) {
                                    ChannelStateWriteRequestExecutor executor =
                                            executorFactory.getOrCreateExecutor(
                                                    jobVertexID,
                                                    i,
                                                    () ->
                                                            CHECKPOINT_STORAGE
                                                                    .createCheckpointStorage(jobID),
                                                    maxSubtasksPerChannelStateFile,
                                                    false);
                                    assertThat(executor).isNotNull();
                                    queue.put(executor);
                                }
                                createFuture.complete(null);
                            } catch (Throwable e) {
                                createFuture.completeExceptionally(e);
                            }
                        })
                .start();

        CompletableFuture<Void> releaseFuture = new CompletableFuture<>();
        new Thread(
                        () -> {
                            try {
                                for (int i = 0; i < numberOfSubtask; i++) {
                                    ChannelStateWriteRequestExecutor executor = queue.take();
                                    executor.releaseSubtask(jobVertexID, numberOfSubtask);
                                }
                                releaseFuture.complete(null);
                            } catch (Throwable e) {
                                releaseFuture.completeExceptionally(e);
                            }
                        })
                .start();

        createFuture.get();
        releaseFuture.get();
    }