@Override
  public long extract(final Object key, final GenericRow value) {
    final String colValue = (String) extractor.extract(key, value);

    try {
      return timestampParser.parse(colValue);
    } catch (final KsqlException e) {
      throw new KsqlException("Unable to parse string timestamp."
          + " timestamp=" + value
          + " timestamp_format=" + format,
          e);
    }
  }
@Test
  public void shouldThrowIfStringDoesNotMatchFormat() {
    // Given:
    when(columnExtractor.extract(any(), any())).thenReturn("11-Jan-2010");

    // When:
    final Exception e = assertThrows(
        DateTimeParseException.class,
        () -> extractor.extract(key, value)
    );

    // Then:
    assertThat(e.getMessage(), containsString("11-Jan-2010"));
  }