code: @SuppressWarnings("unchecked")
  	private <V, A extends Serializable> Accumulator<V, A> getAccumulator(String name,
  			Class<? extends Accumulator<V, A>> accumulatorClass) {
  
  		Accumulator<?, ?> accumulator = accumulators.get(name);
  
  		if (accumulator != null) {
  			AccumulatorHelper.compareAccumulatorTypes(name, accumulator.getClass(), accumulatorClass);
  		} else {
  			// Create new accumulator
  			try {
  				accumulator = accumulatorClass.newInstance();
  			}
  			catch (Exception e) {
  				throw new RuntimeException("Cannot create accumulator " + accumulatorClass.getName());
  			}
  			accumulators.put(name, accumulator);
  		}
  		return (Accumulator<V, A>) accumulator;
  	}
docs: --------------------------------------------------------------------------------------------