code: @Override
  	public void setup(StreamTask<?, ?> containingTask, StreamConfig config, Output<StreamRecord<OUT>> output) {
  		super.setup(containingTask, config, output);
  		FunctionUtils.setFunctionRuntimeContext(userFunction, getRuntimeContext());
  
  	}
docs: ------------------------------------------------------------------------