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 testCombinerOnlyOutputNonBacklogWhenAllInputChannelAreNonBacklog() throws Exception {
        final RecordAttributesCombiner combiner = new RecordAttributesCombiner(2);
        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();

        combiner.inputRecordAttributes(nonBacklogRecordAttribute, 0, collectingDataOutput);
        assertThat(collectingDataOutput.getEvents()).isEmpty();

        combiner.inputRecordAttributes(backlogRecordAttribute, 0, collectingDataOutput);
        combiner.inputRecordAttributes(nonBacklogRecordAttribute, 1, collectingDataOutput);
        assertThat(collectingDataOutput.getEvents()).containsExactly(backlogRecordAttribute);

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