public void inputRecordAttributes(
            RecordAttributes recordAttributes, int channelIdx, DataOutput<?> output)
            throws Exception {
        LOG.debug("RecordAttributes: {} from channel idx: {}", recordAttributes, channelIdx);
        RecordAttributes lastChannelRecordAttributes = allChannelRecordAttributes[channelIdx];
        allChannelRecordAttributes[channelIdx] = recordAttributes;

        // skip if the input RecordAttributes of the input channel is the same as the last.
        if (recordAttributes.equals(lastChannelRecordAttributes)) {
            return;
        }

        final RecordAttributesBuilder builder =
                new RecordAttributesBuilder(Collections.emptyList());

        Boolean isBacklog = combineIsBacklog(lastChannelRecordAttributes, recordAttributes);
        if (isBacklog == null) {
            if (lastOutputAttributes == null) {
                return;
            } else {
                isBacklog = lastOutputAttributes.isBacklog();
            }
        }
        builder.setBacklog(isBacklog);

        final RecordAttributes outputAttribute = builder.build();
        if (!outputAttribute.equals(lastOutputAttributes)) {
            output.emitRecordAttributes(outputAttribute);
            lastOutputAttributes = outputAttribute;
        }
    }
@Test
    void testCombineRecordAttributes() throws Exception {
        final RecordAttributesCombiner combiner = new RecordAttributesCombiner(3);
        CollectingDataOutput<Object> collectingDataOutput = new CollectingDataOutput<>();
        final RecordAttributes backlogRecordAttribute =
                new RecordAttributesBuilder(Collections.emptyList()).setBacklog(true).build();
        final RecordAttributes nonBacklogRecordAttribute =
                new RecordAttributesBuilder(Collections.emptyList()).setBacklog(false).build();

        // Switch from null to backlog
        combiner.inputRecordAttributes(backlogRecordAttribute, 0, collectingDataOutput);
        combiner.inputRecordAttributes(nonBacklogRecordAttribute, 0, collectingDataOutput);
        combiner.inputRecordAttributes(nonBacklogRecordAttribute, 0, collectingDataOutput);

        combiner.inputRecordAttributes(backlogRecordAttribute, 1, collectingDataOutput);
        combiner.inputRecordAttributes(backlogRecordAttribute, 2, collectingDataOutput);

        // Switch from backlog to non-backlog
        combiner.inputRecordAttributes(nonBacklogRecordAttribute, 1, collectingDataOutput);
        combiner.inputRecordAttributes(nonBacklogRecordAttribute, 2, collectingDataOutput);

        // Switch from non-backlog to backlog
        combiner.inputRecordAttributes(backlogRecordAttribute, 1, collectingDataOutput);

        assertThat(collectingDataOutput.getEvents())
                .containsExactly(
                        backlogRecordAttribute, nonBacklogRecordAttribute, backlogRecordAttribute);
    }