@Override
  public long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) {
    try {
      return delegate.extract(record, previousTimestamp);
    } catch (final RuntimeException e) {
      return handleFailure(record.key(), record.value(), e);
    }
  }
@Test
  public void shouldLogExceptionsAndFailOnExtractFromRecord() {
    // Given:
    final LoggingTimestampExtractor extractor = new LoggingTimestampExtractor(
        (k, v) -> {
          throw new KsqlException("foo");
        },
        logger,
        true
    );

    // When/Then:
    assertThrows(
        KsqlException.class,
        () -> extractor.extract(record, PREVIOUS_TS)
    );

    verify(logger).error(any());
  }