public void onBuffer(Buffer buffer, int sequenceNumber, int backlog, int subpartitionId)
            throws IOException {
        boolean recycleBuffer = true;

        try {
            if (expectedSequenceNumber != sequenceNumber) {
                onError(new BufferReorderingException(expectedSequenceNumber, sequenceNumber));
                return;
            }

            if (buffer.getDataType().isBlockingUpstream()) {
                onBlockingUpstream();
                checkArgument(backlog == 0, "Illegal number of backlog: %s, should be 0.", backlog);
            }

            final boolean wasEmpty;
            boolean firstPriorityEvent = false;
            synchronized (receivedBuffers) {
                NetworkActionsLogger.traceInput(
                        "RemoteInputChannel#onBuffer",
                        buffer,
                        inputGate.getOwningTaskName(),
                        channelInfo,
                        channelStatePersister,
                        sequenceNumber);
                // Similar to notifyBufferAvailable(), make sure that we never add a buffer
                // after releaseAllResources() released all buffers from receivedBuffers
                // (see above for details).
                if (isReleased.get()) {
                    return;
                }

                wasEmpty = receivedBuffers.isEmpty();

                SequenceBuffer sequenceBuffer =
                        new SequenceBuffer(buffer, sequenceNumber, subpartitionId);
                DataType dataType = buffer.getDataType();
                if (dataType.hasPriority()) {
                    firstPriorityEvent = addPriorityBuffer(sequenceBuffer);
                    recycleBuffer = false;
                } else {
                    receivedBuffers.add(sequenceBuffer);
                    recycleBuffer = false;
                    if (dataType.requiresAnnouncement()) {
                        firstPriorityEvent = addPriorityBuffer(announce(sequenceBuffer));
                    }
                }
                totalQueueSizeInBytes += buffer.getSize();
                final OptionalLong barrierId =
                        channelStatePersister.checkForBarrier(sequenceBuffer.buffer);
                if (barrierId.isPresent() && barrierId.getAsLong() > lastBarrierId) {
                    // checkpoint was not yet started by task thread,
                    // so remember the numbers of buffers to spill for the time when
                    // it will be started
                    lastBarrierId = barrierId.getAsLong();
                    lastBarrierSequenceNumber = sequenceBuffer.sequenceNumber;
                }
                channelStatePersister.maybePersist(buffer);
                ++expectedSequenceNumber;
            }

            if (firstPriorityEvent) {
                notifyPriorityEvent(sequenceNumber);
            }
            if (wasEmpty) {
                notifyChannelNonEmpty();
            }

            if (backlog >= 0) {
                onSenderBacklog(backlog);
            }
        } finally {
            if (recycleBuffer) {
                buffer.recycleBuffer();
            }
        }
    }
@Test
    void testNotifyOnPriority() throws IOException {
        SingleInputGate inputGate = new SingleInputGateBuilder().build();
        RemoteInputChannel channel = InputChannelTestUtils.createRemoteInputChannel(inputGate, 0);

        CheckpointOptions options = new CheckpointOptions(CHECKPOINT, getDefault());
        assertPriorityAvailability(
                inputGate,
                false,
                false,
                () ->
                        assertAvailability(
                                inputGate,
                                false,
                                true,
                                () -> {
                                    channel.onBuffer(
                                            toBuffer(
                                                    new CheckpointBarrier(1L, 123L, options),
                                                    false),
                                            0,
                                            0,
                                            0);
                                }));
        assertPriorityAvailability(
                inputGate,
                false,
                true,
                () ->
                        assertAvailability(
                                inputGate,
                                true,
                                true,
                                () -> {
                                    channel.onBuffer(
                                            toBuffer(
                                                    new CheckpointBarrier(2L, 123L, options), true),
                                            1,
                                            0,
                                            0);
                                }));
    }