@Test
public void testAddingMetrics() throws NoSuchFieldException, IllegalAccessException {

	Configuration configuration = new Configuration();
	String taskName = "testTask";
	String jobName = "testJob:-!ax..?";
	String hostname = "local::host:";
	String taskManagerId = "tas:kMana::ger";
	String counterName = "testCounter";
	configuration.setString(MetricOptions.REPORTERS_LIST, "test");
	configuration.setString(
			ConfigConstants.METRICS_REPORTER_PREFIX + "test." + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX,
			"org.apache.flink.metrics.statsd.StatsDReporterTest$TestingStatsDReporter");
	configuration.setString(MetricOptions.SCOPE_NAMING_TASK, "<host>.<tm_id>.<job_name>");
	configuration.setString(MetricOptions.SCOPE_DELIMITER, "_");
	MetricRegistryImpl metricRegistry = new MetricRegistryImpl(MetricRegistryConfiguration.fromConfiguration(configuration));
	char delimiter = metricRegistry.getDelimiter();
	TaskManagerMetricGroup tmMetricGroup = new TaskManagerMetricGroup(metricRegistry, hostname, taskManagerId);
	TaskManagerJobMetricGroup tmJobMetricGroup = new TaskManagerJobMetricGroup(metricRegistry, tmMetricGroup, new JobID(), jobName);
	TaskMetricGroup taskMetricGroup = new TaskMetricGroup(metricRegistry, tmJobMetricGroup, new JobVertexID(), new AbstractID(), taskName, 0, 0);
	SimpleCounter myCounter = new SimpleCounter();
	taskMetricGroup.counter(counterName, myCounter);
	List<MetricReporter> reporters = metricRegistry.getReporters();
	assertTrue(reporters.size() == 1);
	MetricReporter metricReporter = reporters.get(0);
	assertTrue("Reporter should be of type StatsDReporter", metricReporter instanceof StatsDReporter);
	TestingStatsDReporter reporter = (TestingStatsDReporter) metricReporter;
	Map<Counter, String> counters = reporter.getCounters();
	assertTrue(counters.containsKey(myCounter));
	String expectedCounterName = reporter.filterCharacters(hostname)
		+ delimiter
		+ reporter.filterCharacters(taskManagerId)
		+ delimiter
		+ reporter.filterCharacters(jobName)
		+ delimiter
		+ reporter.filterCharacters(counterName);
	assertEquals(expectedCounterName, counters.get(myCounter));
	metricRegistry.shutdown();
}