### Question:
AsyncWaitOperator extends AbstractUdfStreamOperator<OUT, AsyncFunction<IN, OUT>> implements OneInputStreamOperator<IN, OUT>, OperatorActions, BoundedOneInput { @Override public void endInput() throws Exception { waitInFlightInputsFinished(); } AsyncWaitOperator(
			AsyncFunction<IN, OUT> asyncFunction,
			long timeout,
			int capacity,
			AsyncDataStream.OutputMode outputMode); @Override void setup(StreamTask<?, ?> containingTask, StreamConfig config, Output<StreamRecord<OUT>> output); @Override void open(); @Override void processElement(StreamRecord<IN> element); @Override void processWatermark(Watermark mark); @Override void snapshotState(StateSnapshotContext context); @Override void initializeState(StateInitializationContext context); @Override void endInput(); @Override void close(); @Override void dispose(); @Override void failOperator(Throwable throwable);  }### Answer:
@Test public void testEndInput() throws Exception { final AsyncWaitOperator<Integer, Integer> operator = new AsyncWaitOperator<>( new DelayedAsyncFunction(10), -1, 2, AsyncDataStream.OutputMode.ORDERED); final OneInputStreamOperatorTestHarness<Integer, Integer> testHarness = new OneInputStreamOperatorTestHarness<>(operator, IntSerializer.INSTANCE); final long initialTime = 0L; final ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>(); expectedOutput.add(new StreamRecord<>(2, initialTime + 1)); expectedOutput.add(new StreamRecord<>(4, initialTime + 2)); expectedOutput.add(new Watermark(initialTime + 2)); expectedOutput.add(new StreamRecord<>(6, initialTime + 3)); testHarness.open(); try { synchronized (testHarness.getCheckpointLock()) { testHarness.processElement(new StreamRecord<>(1, initialTime + 1)); testHarness.processElement(new StreamRecord<>(2, initialTime + 2)); testHarness.processWatermark(new Watermark(initialTime + 2)); testHarness.processElement(new StreamRecord<>(3, initialTime + 3)); } synchronized (testHarness.getCheckpointLock()) { testHarness.endInput(); } TestHarnessUtil.assertOutputEquals("Output with watermark was not correct.", expectedOutput, testHarness.getOutput()); } finally { synchronized (testHarness.getCheckpointLock()) { testHarness.close(); } } }