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

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

        RecordProcessorUtils.getRecordProcessor2(operator1).accept(new StreamRecord<>("test"));
        assertThat(operator1.setKeyContextElement2Called).isTrue();
        assertThat(operator1.processElement2Called).isTrue();

        RecordProcessorUtils.getRecordProcessor2(operator2).accept(new StreamRecord<>("test"));
        assertThat(operator2.setKeyContextElement2Called).isTrue();
        assertThat(operator2.processElement2Called).isTrue();

        RecordProcessorUtils.getRecordProcessor2(operator3).accept(new StreamRecord<>("test"));
        assertThat(operator3.setKeyContextElement2Called).isFalse();
        assertThat(operator3.processElement2Called).isTrue();
    }