@Override
    public void endInput() throws Exception {
        userFunction.endInput(nonPartitionedContext);
    }
@Test
    void testEndInput() throws Exception {
        AtomicInteger counter = new AtomicInteger();
        ProcessOperator<Integer, String> processOperator =
                new ProcessOperator<>(
                        new OneInputStreamProcessFunction<Integer, String>() {
                            @Override
                            public void processRecord(
                                    Integer record,
                                    Collector<String> output,
                                    PartitionedContext ctx) {
                                //  do nothing.
                            }

                            @Override
                            public void endInput(NonPartitionedContext<String> ctx) {
                                try {
                                    ctx.applyToAllPartitions(
                                            (out, context) -> {
                                                counter.incrementAndGet();
                                                out.collect("end");
                                            });

                                } catch (Exception e) {
                                    throw new RuntimeException(e);
                                }
                            }
                        });

        try (OneInputStreamOperatorTestHarness<Integer, String> testHarness =
                new OneInputStreamOperatorTestHarness<>(processOperator)) {
            testHarness.open();
            testHarness.endInput();
            Collection<StreamRecord<String>> recordOutput = testHarness.getRecordOutput();
            assertThat(recordOutput).containsExactly(new StreamRecord<>("end"));
            assertThat(counter).hasValue(1);
        }
    }