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

                            @Override
                            public void processRecordFromSecondInput(
                                    Long record, Collector<Long> output, PartitionedContext ctx) {
                                // do nothing.
                            }

                            @Override
                            public void endFirstInput(NonPartitionedContext<Long> ctx) {
                                try {
                                    ctx.applyToAllPartitions(
                                            (out, context) -> {
                                                firstInputCounter.incrementAndGet();
                                                Long currentKey =
                                                        context.getStateManager().getCurrentKey();
                                                out.collect(currentKey);
                                            });
                                } catch (Exception e) {
                                    throw new RuntimeException(e);
                                }
                            }

                            @Override
                            public void endSecondInput(NonPartitionedContext<Long> ctx) {
                                try {
                                    ctx.applyToAllPartitions(
                                            (out, context) -> {
                                                secondInputCounter.incrementAndGet();
                                                Long currentKey =
                                                        context.getStateManager().getCurrentKey();
                                                out.collect(currentKey);
                                            });
                                } catch (Exception e) {
                                    throw new RuntimeException(e);
                                }
                            }
                        });

        try (KeyedTwoInputStreamOperatorTestHarness<Long, Integer, Long, Long> testHarness =
                new KeyedTwoInputStreamOperatorTestHarness<>(
                        processOperator,
                        (KeySelector<Integer, Long>) Long::valueOf,
                        (KeySelector<Long, Long>) value -> value,
                        Types.LONG)) {
            testHarness.open();
            testHarness.processElement1(new StreamRecord<>(1)); // key is 1L
            testHarness.processElement2(new StreamRecord<>(2L)); // key is 2L
            testHarness.endInput1();
            assertThat(firstInputCounter).hasValue(2);
            Collection<StreamRecord<Long>> recordOutput = testHarness.getRecordOutput();
            assertThat(recordOutput)
                    .containsExactly(new StreamRecord<>(1L), new StreamRecord<>(2L));
            testHarness.processElement2(new StreamRecord<>(3L)); // key is 3L
            testHarness.getOutput().clear();
            testHarness.endInput2();
            assertThat(secondInputCounter).hasValue(3);
            recordOutput = testHarness.getRecordOutput();
            assertThat(recordOutput)
                    .containsExactly(
                            new StreamRecord<>(1L), new StreamRecord<>(2L), new StreamRecord<>(3L));
        }
    }