@Override
    public void processElement1(StreamRecord<IN1> element) throws Exception {
        collector.setTimestamp(element);
        rContext.setElement(element);
        userFunction.processElement(element.getValue(), rContext, collector);
        rContext.setElement(null);
    }
@Test
    void testNoKeyedStateOnNonBroadcastSide() throws Exception {

        boolean exceptionThrown = false;

        final ValueStateDescriptor<String> valueState =
                new ValueStateDescriptor<>("any", BasicTypeInfo.STRING_TYPE_INFO);

        try (TwoInputStreamOperatorTestHarness<String, Integer, String> testHarness =
                getInitializedTestHarness(
                        new BroadcastProcessFunction<String, Integer, String>() {
                            private static final long serialVersionUID = -1725365436500098384L;

                            @Override
                            public void processBroadcastElement(
                                    Integer value, Context ctx, Collector<String> out)
                                    throws Exception {
                                // do nothing
                            }

                            @Override
                            public void processElement(
                                    String value, ReadOnlyContext ctx, Collector<String> out)
                                    throws Exception {
                                assertThatThrownBy(
                                                () ->
                                                        getRuntimeContext()
                                                                .getState(valueState)
                                                                .value())
                                        .isInstanceOf(NullPointerException.class)
                                        .hasMessage(
                                                String.format(
                                                        "Keyed state '%s' with type %s can only be used on a 'keyed stream', i.e., after a 'keyBy()' operation.",
                                                        valueState.getName(),
                                                        valueState.getType()));
                            }
                        })) {
            testHarness.processWatermark1(new Watermark(10L));
            testHarness.processWatermark2(new Watermark(10L));
            testHarness.processElement1(new StreamRecord<>("5", 12L));
        }
    }