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

                            @Override
                            public void processRecordFromSecondInput(
                                    Long record, Collector<Long> output, PartitionedContext ctx) {
                                output.collect(record);
                            }
                        });

        try (KeyedTwoInputStreamOperatorTestHarness<Long, Integer, Long, Long> testHarness =
                new KeyedTwoInputStreamOperatorTestHarness<>(
                        processOperator,
                        (KeySelector<Integer, Long>) (data) -> (long) (data + 1),
                        (KeySelector<Long, Long>) value -> value + 1,
                        Types.LONG)) {
            testHarness.open();
            testHarness.processElement1(new StreamRecord<>(1));
            testHarness.processElement2(new StreamRecord<>(2L));
            testHarness.processElement2(new StreamRecord<>(4L));
            testHarness.processElement1(new StreamRecord<>(3));
            Collection<StreamRecord<Long>> recordOutput = testHarness.getRecordOutput();
            assertThat(recordOutput)
                    .containsExactly(
                            new StreamRecord<>(1L),
                            new StreamRecord<>(2L),
                            new StreamRecord<>(4L),
                            new StreamRecord<>(3L));
        }
    }