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 testOverrideSetKeyContextElementForOneInputStreamOperator() throws Exception {
        // test no override
        NoOverrideOneInputStreamOperator noOverride = new NoOverrideOneInputStreamOperator();
        RecordProcessorUtils.getRecordProcessor(noOverride).accept(new StreamRecord<>("test"));
        assertThat(noOverride.setCurrentKeyCalled).isFalse();

        // test override "SetKeyContextElement"
        OverrideSetKeyContextOneInputStreamOperator overrideSetKeyContext =
                new OverrideSetKeyContextOneInputStreamOperator();
        RecordProcessorUtils.getRecordProcessor(overrideSetKeyContext)
                .accept(new StreamRecord<>("test"));
        assertThat(overrideSetKeyContext.setKeyContextElementCalled).isTrue();

        // test override "SetKeyContextElement1"
        OverrideSetKeyContext1OneInputStreamOperator overrideSetKeyContext1 =
                new OverrideSetKeyContext1OneInputStreamOperator();
        RecordProcessorUtils.getRecordProcessor(overrideSetKeyContext1)
                .accept(new StreamRecord<>("test"));
        assertThat(overrideSetKeyContext1.setKeyContextElement1Called).isTrue();
    }