@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 shouldExtractTimestampFromStringWithFormat() throws ParseException {
    // Given:
    when(columnExtractor.extract(any(), any())).thenReturn("2010-Jan-11");

    // When:
    final long actualTime = extractor.extract(key, value);

    final long expectedTime = new SimpleDateFormat(FORAMT).parse("2010-Jan-11").getTime();
    assertThat(actualTime, equalTo(expectedTime));
  }