public void notifyPartitionRequestTimeout(PartitionRequestListener partitionRequestListener) {
        ctx.pipeline().fireUserEventTriggered(partitionRequestListener);
    }
@Test
    public void testNotifyReaderPartitionTimeout() throws Exception {
        PartitionRequestQueue queue = new PartitionRequestQueue();
        EmbeddedChannel channel = new EmbeddedChannel(queue);
        ResultPartitionManager resultPartitionManager = new ResultPartitionManager();
        ResultPartitionID resultPartitionId = new ResultPartitionID();
        CreditBasedSequenceNumberingViewReader reader =
                new CreditBasedSequenceNumberingViewReader(new InputChannelID(0, 0), 10, queue);
        reader.requestSubpartitionViewOrRegisterListener(
                resultPartitionManager, resultPartitionId, new ResultSubpartitionIndexSet(0));

        assertThat(
                        resultPartitionManager
                                .getListenerManagers()
                                .get(resultPartitionId)
                                .getPartitionRequestListeners())
                .hasSize(1);

        reader.notifyPartitionRequestTimeout(
                resultPartitionManager
                        .getListenerManagers()
                        .get(resultPartitionId)
                        .getPartitionRequestListeners()
                        .iterator()
                        .next());

        channel.runPendingTasks();

        Object read = channel.readOutbound();
        assertThat(read)
                .isNotNull()
                .isInstanceOf(NettyMessage.ErrorResponse.class)
                .isInstanceOfSatisfying(
                        NettyMessage.ErrorResponse.class,
                        r -> assertThat(r.cause).isInstanceOf(PartitionNotFoundException.class));
    }