@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 counter = new AtomicInteger();
        OutputTag<Long> sideOutputTag = new OutputTag<Long>("side-output") {};

        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) {
                                // do nothing.
                            }

                            @Override
                            public void endInput(
                                    TwoOutputNonPartitionedContext<Integer, Long> ctx) {
                                try {
                                    ctx.applyToAllPartitions(
                                            (firstOutput, secondOutput, context) -> {
                                                counter.incrementAndGet();
                                                Integer currentKey =
                                                        context.getStateManager().getCurrentKey();
                                                firstOutput.collect(currentKey);
                                                secondOutput.collect(Long.valueOf(currentKey));
                                            });
                                } catch (Exception e) {
                                    throw new RuntimeException(e);
                                }
                            }
                        },
                        sideOutputTag);

        try (KeyedOneInputStreamOperatorTestHarness<Integer, Integer, Integer> testHarness =
                new KeyedOneInputStreamOperatorTestHarness<>(
                        processOperator,
                        (KeySelector<Integer, Integer>) value -> value,
                        Types.INT)) {
            testHarness.open();
            testHarness.processElement(new StreamRecord<>(1)); // key is 1
            testHarness.processElement(new StreamRecord<>(2)); //  key is 2
            testHarness.endInput();
            assertThat(counter).hasValue(2);
            Collection<StreamRecord<Integer>> firstOutput = testHarness.getRecordOutput();
            ConcurrentLinkedQueue<StreamRecord<Long>> secondOutput =
                    testHarness.getSideOutput(sideOutputTag);
            assertThat(firstOutput).containsExactly(new StreamRecord<>(1), new StreamRecord<>(2));
            assertThat(secondOutput)
                    .containsExactly(new StreamRecord<>(1L), new StreamRecord<>(2L));
        }
    }