@Override
    public void open() throws Exception {
        this.timerService =
                getInternalTimerService("processing timer", VoidNamespaceSerializer.INSTANCE, this);
        this.keySet = new HashSet<>();
        super.open();
    }
@Test
    void testCheckKey() throws Exception {
        KeyedTwoInputBroadcastProcessOperator<Long, Integer, Long, Long> processOperator =
                new KeyedTwoInputBroadcastProcessOperator<>(
                        new TwoInputBroadcastStreamProcessFunction<Integer, Long, Long>() {
                            @Override
                            public void processRecordFromNonBroadcastInput(
                                    Integer record,
                                    Collector<Long> output,
                                    PartitionedContext ctx) {
                                output.collect(Long.valueOf(record));
                            }

                            @Override
                            public void processRecordFromBroadcastInput(
                                    Long record, NonPartitionedContext<Long> ctx) throws Exception {
                                ctx.applyToAllPartitions(
                                        (collect, context) -> collect.collect(record));
                            }
                        },
                        // -1 is an invalid key in this suite.
                        (out) -> -1L);

        try (KeyedTwoInputStreamOperatorTestHarness<Long, Integer, Long, Long> testHarness =
                new KeyedTwoInputStreamOperatorTestHarness<>(
                        processOperator,
                        (KeySelector<Integer, Long>) Long::valueOf,
                        (KeySelector<Long, Long>) value -> value,
                        Types.LONG)) {
            testHarness.open();
            assertThatThrownBy(() -> testHarness.processElement1(new StreamRecord<>(1)))
                    .isInstanceOf(IllegalStateException.class);
            // TODO also test processElement2 after non-partitioned context fully implemented.
        }
    }