@Override
    public Optional<BufferAndAvailability> getNextBuffer() throws IOException {
        checkPartitionRequestQueueInitialized();

        final SequenceBuffer next;
        final DataType nextDataType;

        synchronized (receivedBuffers) {
            next = receivedBuffers.poll();

            if (next != null) {
                totalQueueSizeInBytes -= next.buffer.getSize();
            }
            nextDataType =
                    receivedBuffers.peek() != null
                            ? receivedBuffers.peek().buffer.getDataType()
                            : DataType.NONE;
        }

        if (next == null) {
            if (isReleased.get()) {
                throw new CancelTaskException(
                        "Queried for a buffer after channel has been released.");
            }
            return Optional.empty();
        }

        NetworkActionsLogger.traceInput(
                "RemoteInputChannel#getNextBuffer",
                next.buffer,
                inputGate.getOwningTaskName(),
                channelInfo,
                channelStatePersister,
                next.sequenceNumber);
        numBytesIn.inc(next.buffer.getSize());
        numBuffersIn.inc();
        return Optional.of(
                new BufferAndAvailability(next.buffer, nextDataType, 0, next.sequenceNumber));
    }
@Test
    void testRequiresAnnouncement() throws Exception {
        int sequenceNumber = 0;
        int bufferSize = 1;
        final RemoteInputChannel channel = buildInputGateAndGetChannel(sequenceNumber);
        sendBuffer(channel, sequenceNumber++, bufferSize++);
        sendBuffer(channel, sequenceNumber++, bufferSize++);
        sendBarrier(channel, sequenceNumber++, ALIGNED_WITH_TIMEOUT);
        sendBuffer(channel, sequenceNumber++, bufferSize++);
        sendBuffer(channel, sequenceNumber++, bufferSize++);

        BufferAndAvailability nextBuffer = channel.getNextBuffer().get();
        assertThat(nextBuffer.getSequenceNumber()).isEqualTo(2);
        assertThat(nextBuffer.morePriorityEvents()).isFalse();
        assertThat(nextBuffer.moreAvailable()).isTrue();
        assertThat(nextBuffer.buffer().getDataType()).isEqualTo(DataType.PRIORITIZED_EVENT_BUFFER);

        assertGetNextBufferSequenceNumbers(channel, 0, 1);

        nextBuffer = channel.getNextBuffer().get();
        assertThat(nextBuffer.getSequenceNumber()).isEqualTo(2);
        assertThat(nextBuffer.buffer().getDataType())
                .isEqualTo(DataType.TIMEOUTABLE_ALIGNED_CHECKPOINT_BARRIER);

        assertThat(channel.getNextBuffer().get().getSequenceNumber()).isEqualTo(3);
    }