public void finish(StreamTaskActionExecutor actionExecutor, StopMode stopMode)
            throws Exception {
        if (!isHead && stopMode == StopMode.DRAIN) {
            // NOTE: This only do for the case where the operator is one-input operator. At present,
            // any non-head operator on the operator chain is one-input operator.
            actionExecutor.runThrowing(() -> endOperatorInput(1));
        }

        quiesceTimeServiceAndFinishOperator(actionExecutor, stopMode);

        // propagate the close operation to the next wrapper
        if (next != null) {
            next.finish(actionExecutor, stopMode);
        }
    }
@Test
    void testFinish() throws Exception {
        output.clear();
        operatorWrappers.get(0).finish(containingTask.getActionExecutor(), StopMode.DRAIN);

        List<Object> expected = new ArrayList<>();
        for (int i = 0; i < operatorWrappers.size(); i++) {
            String prefix = "[" + "Operator" + i + "]";
            Collections.addAll(
                    expected,
                    prefix + ": End of input",
                    prefix + ": Timer that was in mailbox before closing operator",
                    prefix + ": Bye",
                    prefix + ": Mail to put in mailbox when finishing operator");
        }

        assertThat(output)
                .as("Output was not correct.")
                .containsExactlyElementsOf(expected.subList(2, expected.size()));
    }