### Question:
AvroSimpleKeyExample implements SimpleUDF { @Override public void open(Configuration parameters, RuntimeContext runtimeContext) { getKeysDuration = runtimeContext.getMetricGroup().counter("getKeysDuration"); aggregateDuration = runtimeContext.getMetricGroup().counter("aggregateDuration"); }  @Override void open(Configuration parameters, RuntimeContext runtimeContext); @Override Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] payload); @Override byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages); @Override byte[] merge(Map<String, Object> keys, Iterator<byte[]> files);  }### Answer:
@Test public void testRegisterMetrics() { example.open(parameters, runtimeContext); verify(runtimeContext, times(2)).getMetricGroup(); verify(metricGroup, times(1)).counter(eq("getKeysDuration")); verify(metricGroup, times(1)).counter(eq("aggregateDuration")); }