@Override
    protected void requestSubpartitions() throws IOException {

        boolean retriggerRequest = false;
        boolean notifyDataAvailable = false;

        // The lock is required to request only once in the presence of retriggered requests.
        synchronized (requestLock) {
            checkState(!isReleased, "LocalInputChannel has been released already");

            if (subpartitionView == null) {
                LOG.debug(
                        "{}: Requesting LOCAL subpartitions {} of partition {}. {}",
                        this,
                        consumedSubpartitionIndexSet,
                        partitionId,
                        channelStatePersister);

                try {
                    ResultSubpartitionView subpartitionView =
                            partitionManager.createSubpartitionView(
                                    partitionId, consumedSubpartitionIndexSet, this);

                    if (subpartitionView == null) {
                        throw new IOException("Error requesting subpartition.");
                    }

                    // make the subpartition view visible
                    this.subpartitionView = subpartitionView;

                    // check if the channel was released in the meantime
                    if (isReleased) {
                        subpartitionView.releaseAllResources();
                        this.subpartitionView = null;
                    } else {
                        notifyDataAvailable = true;
                    }
                } catch (PartitionNotFoundException notFound) {
                    if (increaseBackoff()) {
                        retriggerRequest = true;
                    } else {
                        throw notFound;
                    }
                }
            }
        }

        if (notifyDataAvailable) {
            notifyDataAvailable(this.subpartitionView);
        }

        // Do this outside of the lock scope as this might lead to a
        // deadlock with a concurrent release of the channel via the
        // input gate.
        if (retriggerRequest) {
            inputGate.retriggerPartitionRequest(partitionId.getPartitionId(), channelInfo);
        }
    }
@Test
    void testConcurrentReleaseAndRetriggerPartitionRequest() throws Exception {
        final SingleInputGate gate = createSingleInputGate(1);

        ResultPartitionManager partitionManager = mock(ResultPartitionManager.class);
        when(partitionManager.createSubpartitionView(
                        any(ResultPartitionID.class),
                        any(ResultSubpartitionIndexSet.class),
                        any(BufferAvailabilityListener.class)))
                .thenAnswer(
                        (Answer<ResultSubpartitionView>)
                                invocationOnMock -> {
                                    // Sleep here a little to give the releaser Thread
                                    // time to acquire the input gate lock. We throw
                                    // the Exception to retrigger the request.
                                    Thread.sleep(100);
                                    throw new PartitionNotFoundException(new ResultPartitionID());
                                });

        final LocalInputChannel channel = createLocalInputChannel(gate, partitionManager, 1, 1);

        Thread releaser =
                new Thread(
                        () -> {
                            try {
                                gate.close();
                            } catch (IOException ignored) {
                            }
                        });

        Thread requester =
                new Thread(
                        () -> {
                            try {
                                channel.requestSubpartitions();
                            } catch (IOException ignored) {
                            }
                        });

        requester.start();
        releaser.start();

        releaser.join();
        requester.join();
    }