@Override
    public void open() throws Exception {
        this.timerService =
                getInternalTimerService("processing timer", VoidNamespaceSerializer.INSTANCE, this);
        this.keySet = new HashSet<>();
        super.open();
    }
@Test
    void testKeyCheck() throws Exception {
        OutputTag<Long> sideOutputTag = new OutputTag<Long>("side-output") {};
        AtomicBoolean emitToFirstOutput = new AtomicBoolean(true);
        KeyedTwoOutputProcessOperator<Integer, Integer, Integer, Long> processOperator =
                new KeyedTwoOutputProcessOperator<>(
                        new TwoOutputStreamProcessFunction<Integer, Integer, Long>() {
                            @Override
                            public void processRecord(
                                    Integer record,
                                    Collector<Integer> output1,
                                    Collector<Long> output2,
                                    PartitionedContext ctx) {
                                if (emitToFirstOutput.get()) {
                                    output1.collect(record);
                                } else {
                                    output2.collect((long) (record));
                                }
                            }
                        },
                        sideOutputTag,
                        // -1 is an invalid key in this suite.
                        (KeySelector<Integer, Integer>) value -> -1,
                        // -1 is an invalid key in this suite.
                        (KeySelector<Long, Integer>) value -> -1);

        try (KeyedOneInputStreamOperatorTestHarness<Integer, Integer, Integer> testHarness =
                new KeyedOneInputStreamOperatorTestHarness<>(
                        processOperator,
                        (KeySelector<Integer, Integer>) value -> value,
                        Types.INT)) {
            testHarness.open();
            assertThatThrownBy(() -> testHarness.processElement(new StreamRecord<>(1)))
                    .isInstanceOf(IllegalStateException.class);
            emitToFirstOutput.set(false);
            assertThatThrownBy(() -> testHarness.processElement(new StreamRecord<>(1)))
                    .isInstanceOf(IllegalStateException.class);
        }
    }