code: private void openAllOperators() throws Exception {
  		for (StreamOperator<?> operator : operatorChain.getAllOperators()) {
  			if (operator != null) {
  				operator.open();
  			}
  		}
  	}
docs: Execute {@link StreamOperator#open()} of each operator in the chain of this
  {@link StreamTask}. Opening happens from <b>tail to head</b> operator in the chain, contrary
  to {@link StreamOperator#close()} which happens <b>head to tail</b>
  (see {@link #closeAllOperators()}.