class JdbcSource implements SupportedConnector {

  @Override
  public Map<String, String> resolveConfigs(final Map<String, String> configs) {
    final Map<String, String> resolved = new HashMap<>(configs);
    final String key = resolved.remove("key");
    if (key != null) {
      resolved.merge(
          "transforms",
          "ksqlCreateKey,ksqlExtractString",
          (a, b) -> String.join(",", a, b));

      resolved.put("transforms.ksqlCreateKey.type", ValueToKey.class.getName());
      resolved.put("transforms.ksqlCreateKey.fields", key);

      resolved.put("transforms.ksqlExtractString.type", ExtractField.Key.class.getName());
      resolved.put("transforms.ksqlExtractString.field", key);
    }

    resolved.putIfAbsent("tasks.max", "1");
    return resolved;
  }

}