class StreamOperatorWrapper {

	public void close(StreamTaskActionExecutor actionExecutor) throws Exception {
		if (!isHead) {
			// 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));
		}

		quiesceTimeServiceAndCloseOperator(actionExecutor);

		// propagate the close operation to the next wrapper
		if (next != null) {
			next.close(actionExecutor);
		}
	}

	  StreamOperatorWrapper(
			OP wrapped,
			Optional<ProcessingTimeService> processingTimeService,
			MailboxExecutor mailboxExecutor,
			boolean isHead);

}