code: public void updateAccumulators(AccumulatorSnapshot accumulatorSnapshot) {
  		Map<String, Accumulator<?, ?>> userAccumulators;
  		try {
  			userAccumulators = accumulatorSnapshot.deserializeUserAccumulators(userClassLoader);
  
  			ExecutionAttemptID execID = accumulatorSnapshot.getExecutionAttemptID();
  			Execution execution = currentExecutions.get(execID);
  			if (execution != null) {
  				execution.setAccumulators(userAccumulators);
  			} else {
  				LOG.debug("Received accumulator result for unknown execution {}.", execID);
  			}
  		} catch (Exception e) {
  			LOG.error("Cannot update accumulators for job {}.", getJobID(), e);
  		}
  	}
docs: Updates the accumulators during the runtime of a job. Final accumulator results are transferred
  through the UpdateTaskExecutionState message.
  @param accumulatorSnapshot The serialized flink and user-defined accumulators