public static <T> ThrowingConsumer<StreamRecord<T>, Exception> getRecordProcessor(
            Input<T> input) {
        boolean canOmitSetKeyContext;
        if (input instanceof AbstractStreamOperator) {
            canOmitSetKeyContext = canOmitSetKeyContext((AbstractStreamOperator<?>) input, 0);
        } else {
            canOmitSetKeyContext =
                    input instanceof KeyContextHandler
                            && !((KeyContextHandler) input).hasKeyContext();
        }

        if (canOmitSetKeyContext) {
            return input::processElement;
        } else if (input instanceof AsyncStateProcessing
                && ((AsyncStateProcessing) input).isAsyncStateProcessingEnabled()) {
            return ((AsyncStateProcessing) input).getRecordProcessor(1);
        } else {
            return record -> {
                input.setKeyContextElement(record);
                input.processElement(record);
            };
        }
    }
@Test
    void testGetRecordProcessor() throws Exception {
        TestOperator input1 = new TestOperator();
        TestOperator input2 = new TestKeyContextHandlerOperator(true);
        TestOperator input3 = new TestKeyContextHandlerOperator(false);

        RecordProcessorUtils.getRecordProcessor(input1).accept(new StreamRecord<>("test"));
        assertThat(input1.setKeyContextElementCalled).isTrue();
        assertThat(input1.processElementCalled).isTrue();

        RecordProcessorUtils.getRecordProcessor(input2).accept(new StreamRecord<>("test"));
        assertThat(input2.setKeyContextElementCalled).isTrue();
        assertThat(input2.processElementCalled).isTrue();

        RecordProcessorUtils.getRecordProcessor(input3).accept(new StreamRecord<>("test"));
        assertThat(input3.setKeyContextElementCalled).isFalse();
        assertThat(input3.processElementCalled).isTrue();
    }