@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 nonBroadcastInputCounter = new AtomicInteger();
        AtomicInteger broadcastInputCounter = new AtomicInteger();
        KeyedTwoInputBroadcastProcessOperator<Long, Integer, Long, Long> processOperator =
                new KeyedTwoInputBroadcastProcessOperator<>(
                        new TwoInputBroadcastStreamProcessFunction<Integer, Long, Long>() {
                            @Override
                            public void processRecordFromNonBroadcastInput(
                                    Integer record,
                                    Collector<Long> output,
                                    PartitionedContext ctx) {
                                //  do nothing.
                            }

                            @Override
                            public void processRecordFromBroadcastInput(
                                    Long record, NonPartitionedContext<Long> ctx) {
                                // do nothing.
                            }

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

                            @Override
                            public void endBroadcastInput(NonPartitionedContext<Long> ctx) {
                                try {
                                    ctx.applyToAllPartitions(
                                            (out, context) -> {
                                                broadcastInputCounter.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)); // broadcast input is not keyed
            testHarness.endInput1();
            assertThat(nonBroadcastInputCounter).hasValue(1);
            Collection<StreamRecord<Long>> recordOutput = testHarness.getRecordOutput();
            assertThat(recordOutput).containsExactly(new StreamRecord<>(1L));
            testHarness.processElement2(new StreamRecord<>(3L)); // broadcast input is not keyed
            testHarness.endInput2();
            assertThat(broadcastInputCounter).hasValue(1);
            recordOutput = testHarness.getRecordOutput();
            assertThat(recordOutput)
                    .containsExactly(new StreamRecord<>(1L), new StreamRecord<>(1L));
        }
    }