code: private void closeAllOperators() throws Exception {
  		// We need to close them first to last, since upstream operators in the chain might emit
  		// elements in their close methods.
  		StreamOperator<?>[] allOperators = operatorChain.getAllOperators();
  		for (int i = allOperators.length - 1; i >= 0; i--) {
  			StreamOperator<?> operator = allOperators[i];
  			if (operator != null) {
  				operator.close();
  			}
  		}
  	}
docs: Execute {@link StreamOperator#close()} of each operator in the chain of this
  {@link StreamTask}. Closing happens from <b>head to tail</b> operator in the chain,
  contrary to {@link StreamOperator#open()} which happens <b>tail to head</b>
  (see {@link #openAllOperators()}.