class 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);
    }
  }

}