@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 shouldLogExceptionsAndNotFailOnExtractFromRecord() {
    // Given:
    final KsqlException e = new KsqlException("foo");
    final LoggingTimestampExtractor extractor = new LoggingTimestampExtractor(
        (k, v) -> {
          throw e;
        },
        logger,
        false
    );

    // When:
    final long result = extractor.extract(record, PREVIOUS_TS);

    // Then (did not throw):
    verify(logger).error(RecordProcessingError
        .recordProcessingError("Failed to extract timestamp from row", e,
            () -> "key:" + KEY + ", value:" + VALUE));

    assertThat(result, is(-1L));
  }