public void onFailedPartitionRequest() {
        inputGate.triggerPartitionStateCheck(partitionId, channelInfo);
    }
@Test
    void testOnFailedPartitionRequestDoesNotBlockNetworkThreads() throws Exception {

        final long testBlockedWaitTimeoutMillis = 30_000L;

        final PartitionProducerStateChecker partitionProducerStateChecker =
                (jobId, intermediateDataSetId, resultPartitionId) ->
                        CompletableFuture.completedFuture(ExecutionState.RUNNING);
        final NettyShuffleEnvironment shuffleEnvironment =
                new NettyShuffleEnvironmentBuilder().build();
        final Task task =
                new TestTaskBuilder(shuffleEnvironment)
                        .setPartitionProducerStateChecker(partitionProducerStateChecker)
                        .build(EXECUTOR_EXTENSION.getExecutor());
        final SingleInputGate inputGate =
                new SingleInputGateBuilder().setPartitionProducerStateProvider(task).build();

        TestTaskBuilder.setTaskState(task, ExecutionState.RUNNING);

        final OneShotLatch ready = new OneShotLatch();
        final OneShotLatch blocker = new OneShotLatch();
        final AtomicBoolean timedOutOrInterrupted = new AtomicBoolean(false);

        final ConnectionManager blockingConnectionManager =
                new TestingConnectionManager() {

                    @Override
                    public PartitionRequestClient createPartitionRequestClient(
                            ConnectionID connectionId) {
                        ready.trigger();
                        try {
                            // We block here, in a section that holds the
                            // SingleInputGate#requestLock
                            blocker.await(testBlockedWaitTimeoutMillis, TimeUnit.MILLISECONDS);
                        } catch (InterruptedException | TimeoutException e) {
                            timedOutOrInterrupted.set(true);
                        }

                        return new TestingPartitionRequestClient();
                    }
                };

        final RemoteInputChannel remoteInputChannel =
                InputChannelBuilder.newBuilder()
                        .setConnectionManager(blockingConnectionManager)
                        .buildRemoteChannel(inputGate);
        inputGate.setInputChannels(remoteInputChannel);

        final Thread simulatedNetworkThread =
                new Thread(
                        () -> {
                            try {
                                ready.await();
                                // We want to make sure that our simulated network thread does not
                                // block on
                                // SingleInputGate#requestLock as well through this call.
                                remoteInputChannel.onFailedPartitionRequest();

                                // Will only give free the blocker if we did not block ourselves.
                                blocker.trigger();
                            } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                            }
                        });

        simulatedNetworkThread.start();

        // The entry point to that will lead us into
        // blockingConnectionManager#createPartitionRequestClient(...).
        inputGate.requestPartitions();

        simulatedNetworkThread.join();

        assertThat(timedOutOrInterrupted)
                .withFailMessage(
                        "Test ended by timeout or interruption - this indicates that the network thread was blocked.")
                .isFalse();
    }