private void announceBacklog(NetworkSequenceViewReader reader, int backlog) {
        checkArgument(backlog > 0, "Backlog must be positive.");

        NettyMessage.BacklogAnnouncement announcement =
                new NettyMessage.BacklogAnnouncement(backlog, reader.getReceiverId());
        ctx.channel()
                .writeAndFlush(announcement)
                .addListener(
                        (ChannelFutureListener)
                                future -> {
                                    if (!future.isSuccess()) {
                                        onChannelFutureFailure(future);
                                    }
                                });
    }
@Test
    void testAnnounceBacklog() throws Exception {
        PipelinedSubpartition subpartition =
                PipelinedSubpartitionTest.createPipelinedSubpartition();
        subpartition.add(createEventBufferConsumer(4096, Buffer.DataType.DATA_BUFFER));
        subpartition.add(createEventBufferConsumer(4096, Buffer.DataType.DATA_BUFFER));

        PipelinedSubpartitionView view =
                subpartition.createReadView(new NoOpBufferAvailablityListener());
        ResultPartition partition =
                TestingResultPartition.newBuilder()
                        .setCreateSubpartitionViewFunction((index, listener) -> view)
                        .build();

        PartitionRequestQueue queue = new PartitionRequestQueue();
        InputChannelID receiverId = new InputChannelID();
        CreditBasedSequenceNumberingViewReader reader =
                new CreditBasedSequenceNumberingViewReader(receiverId, 0, queue);
        EmbeddedChannel channel = new EmbeddedChannel(queue);

        reader.notifySubpartitionsCreated(partition, new ResultSubpartitionIndexSet(0));
        queue.notifyReaderCreated(reader);

        reader.notifyDataAvailable(view);
        channel.runPendingTasks();
        Object data = channel.readOutbound();
        assertThat(data).isInstanceOf(NettyMessage.BacklogAnnouncement.class);
        NettyMessage.BacklogAnnouncement announcement = (NettyMessage.BacklogAnnouncement) data;
        assertThat(announcement.receiverId).isEqualTo(receiverId);
        assertThat(announcement.backlog).isEqualTo(subpartition.getBuffersInBacklogUnsafe());

        subpartition.release();
        reader.notifyDataAvailable(view);
        channel.runPendingTasks();
        assertThat((Object) channel.readOutbound()).isNotNull();
    }