docs: @Override
  	public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
  		final String fullName = group.getMetricIdentifier(metricName, this);
  
  		synchronized (this) {
  			if (metric instanceof Counter) {
  				counters.put((Counter) metric, fullName);
  				registry.register(fullName, new FlinkCounterWrapper((Counter) metric));
  			}
  			else if (metric instanceof Gauge) {
  				gauges.put((Gauge<?>) metric, fullName);
  				registry.register(fullName, FlinkGaugeWrapper.fromGauge((Gauge<?>) metric));
  			} else if (metric instanceof Histogram) {
  				Histogram histogram = (Histogram) metric;
  				histograms.put(histogram, fullName);
  
  				if (histogram instanceof DropwizardHistogramWrapper) {
  					registry.register(fullName, ((DropwizardHistogramWrapper) histogram).getDropwizardHistogram());
  				} else {
  					registry.register(fullName, new FlinkHistogramWrapper(histogram));
  				}
  			} else if (metric instanceof Meter) {
  				Meter meter = (Meter) metric;
  				meters.put(meter, fullName);
  
  				if (meter instanceof DropwizardMeterWrapper) {
  					registry.register(fullName, ((DropwizardMeterWrapper) meter).getDropwizardMeter());
  				} else {
  					registry.register(fullName, new FlinkMeterWrapper(meter));
  				}
  			} else {
  				log.warn("Cannot add metric of type {}. This indicates that the reporter " +
  					"does not support this metric type.", metric.getClass().getName());
  			}
  		}
  	}