@Override
  public Object evaluate(final Object... args) {
    final long start = time.nanoseconds();
    try {
      return kudf.evaluate(args);
    } finally {
      sensor.record(time.nanoseconds() - start);
    }
  }
@Test
  public void shouldRecordEvenIfExceptionThrown(){
    final UdfMetricProducer metricProducer
        = new UdfMetricProducer(sensor, args -> {
          time.sleep(10);
     throw new RuntimeException("boom");
    }, time);

    try {
      metricProducer.evaluate("foo");
    } catch (final Exception e) {
      // ignored
    }

    final KafkaMetric metric = metrics.metric(metricName);
    final Double actual = (Double) metric.metricValue();
    assertThat(actual.longValue(), equalTo(TimeUnit.MILLISECONDS.toNanos(10)));
  }