### Question:
StringTimestampExtractor implements TimestampExtractor, Configurable { @Override public void configure(final Map<String, ?> map) { format = (String) map.get(KsqlConfig.STRING_TIMESTAMP_FORMAT); if (format == null) { throw new ConfigException("Value of " + KsqlConfig.STRING_TIMESTAMP_FORMAT + " must not be null"); } final Integer index = (Integer) map.get(KsqlConfig.KSQL_TIMESTAMP_COLUMN_INDEX); if (index == null || index < 0) { throw new ConfigException("Value of " + KsqlConfig.KSQL_TIMESTAMP_COLUMN_INDEX + " must be an integer >= 0"); } this.timestampColumn = index; try { this.timestampParser = new StringToTimestampParser(format); } catch (Exception e) { throw new ConfigException("Invalid date format: " + format + " " + e.getMessage(), e); } }  @Override long extract(final ConsumerRecord<Object, Object> consumerRecord,
                      final long previousTimestamp); @Override void configure(final Map<String, ?> map);  }### Answer:
@Test(expected = ConfigException.class) public void shouldThrowIfFormatNotSuppliedDuringConfigure() { timestampExtractor.configure(Collections.singletonMap( KsqlConfig.KSQL_TIMESTAMP_COLUMN_INDEX, 0)); }
@Test(expected = ConfigException.class) public void shouldThrowIfColumnIndexNotSet() { timestampExtractor.configure(Collections.singletonMap( KsqlConfig.STRING_TIMESTAMP_FORMAT, format)); }
@Test(expected = ConfigException.class) public void shouldThrowIfColumnIndexIsNegative() { final Map props = new HashMap() {{ put(KsqlConfig.STRING_TIMESTAMP_FORMAT, format); put(KsqlConfig.KSQL_TIMESTAMP_COLUMN_INDEX, -1); }}; timestampExtractor.configure(props); }
@SuppressWarnings("unchecked") @Test(expected = ConfigException.class) public void shouldThrowOnInvalidFormat() { final Map props = new HashMap() {{ put(KsqlConfig.STRING_TIMESTAMP_FORMAT, "lahdfl"); put(KsqlConfig.KSQL_TIMESTAMP_COLUMN_INDEX, 0); }}; timestampExtractor.configure(props); }