@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;
  }
@Test
  public void shouldResolveJdbcSourceConfigsTemplate() {
    // Given:
    final Map<String, String> originals = ImmutableMap.<String, String>builder()
        .put(Connectors.CONNECTOR_CLASS, JdbcSource.JDBC_SOURCE_CLASS)
        .put("transforms", "foo")
        .put("key", "id")
        .build();

    // When:
    final Map<String, String> resolved = jdbcSource.resolveConfigs(originals);

    // Then:
    assertThat(
        resolved,
        is(ImmutableMap.<String, String>builder()
            .put(Connectors.CONNECTOR_CLASS, JdbcSource.JDBC_SOURCE_CLASS)
            .put("transforms", "foo,ksqlCreateKey,ksqlExtractString")
            .put("transforms.ksqlCreateKey.type", "org.apache.kafka.connect.transforms.ValueToKey")
            .put("transforms.ksqlCreateKey.fields", "id")
            .put("transforms.ksqlExtractString.type", "org.apache.kafka.connect.transforms.ExtractField$Key")
            .put("transforms.ksqlExtractString.field", "id")
            .put("tasks.max", "1")
            .build()));
  }