public static <T> ThrowingConsumer<StreamRecord<T>, Exception> getRecordProcessor1(
            TwoInputStreamOperator<T, ?, ?> operator) {
        boolean canOmitSetKeyContext;
        if (operator instanceof AbstractStreamOperator) {
            canOmitSetKeyContext = canOmitSetKeyContext((AbstractStreamOperator<?>) operator, 0);
        } else {
            canOmitSetKeyContext =
                    operator instanceof KeyContextHandler
                            && !((KeyContextHandler) operator).hasKeyContext1();
        }

        if (canOmitSetKeyContext) {
            return operator::processElement1;
        } else if (operator instanceof AsyncStateProcessing
                && ((AsyncStateProcessing) operator).isAsyncStateProcessingEnabled()) {
            return ((AsyncStateProcessing) operator).getRecordProcessor(1);
        } else {
            return record -> {
                operator.setKeyContextElement1(record);
                operator.processElement1(record);
            };
        }
    }
@Test
    void testGetRecordProcessor1() throws Exception {
        TestOperator operator1 = new TestOperator();
        TestOperator operator2 = new TestKeyContextHandlerOperator(true, true);
        TestOperator operator3 = new TestKeyContextHandlerOperator(false, true);

        RecordProcessorUtils.getRecordProcessor1(operator1).accept(new StreamRecord<>("test"));
        assertThat(operator1.setKeyContextElement1Called).isTrue();
        assertThat(operator1.processElement1Called).isTrue();

        RecordProcessorUtils.getRecordProcessor1(operator2).accept(new StreamRecord<>("test"));
        assertThat(operator2.setKeyContextElement1Called).isTrue();
        assertThat(operator2.processElement1Called).isTrue();

        RecordProcessorUtils.getRecordProcessor1(operator3).accept(new StreamRecord<>("test"));
        assertThat(operator3.setKeyContextElement1Called).isFalse();
        assertThat(operator3.processElement1Called).isTrue();
    }