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 testFinishingOperatorWithException() {
        AbstractStreamOperator<Void> streamOperator =
                new AbstractStreamOperator<Void>() {
                    @Override
                    public void finish() throws Exception {
                        throw new Exception("test exception at finishing");
                    }
                };

        StreamOperatorWrapper<?, ?> operatorWrapper =
                new StreamOperatorWrapper<>(
                        streamOperator,
                        Optional.ofNullable(streamOperator.getProcessingTimeService()),
                        containingTask
                                .getMailboxExecutorFactory()
                                .createExecutor(Integer.MAX_VALUE - 1),
                        true);

        assertThatThrownBy(
                        () ->
                                operatorWrapper.finish(
                                        containingTask.getActionExecutor(), StopMode.DRAIN))
                .hasMessageContaining("test exception at finishing");
    }