@Test
public void testTaskIOMetricGroup() {

	TaskMetricGroup task = new UnregisteredTaskMetricsGroup();
	TaskIOMetricGroup taskIO = task.getIOMetricGroup();
	// test counter forwarding
	assertNotNull(taskIO.getNumRecordsInCounter());
	assertNotNull(taskIO.getNumRecordsOutCounter());
	Counter c1 = new SimpleCounter();
	c1.inc(32L);
	Counter c2 = new SimpleCounter();
	c2.inc(64L);
	taskIO.reuseRecordsInputCounter(c1);
	taskIO.reuseRecordsOutputCounter(c2);
	assertEquals(32L, taskIO.getNumRecordsInCounter().getCount());
	assertEquals(64L, taskIO.getNumRecordsOutCounter().getCount());
	// test IOMetrics instantiation
	taskIO.getNumBytesInLocalCounter().inc(100L);
	taskIO.getNumBytesInRemoteCounter().inc(150L);
	taskIO.getNumBytesOutCounter().inc(250L);
	IOMetrics io = taskIO.createSnapshot();
	assertEquals(32L, io.getNumRecordsIn());
	assertEquals(64L, io.getNumRecordsOut());
	assertEquals(100L, io.getNumBytesInLocal());
	assertEquals(150L, io.getNumBytesInRemote());
	assertEquals(250L, io.getNumBytesOut());
}