@VisibleForTesting
  Optional<Connector> fromConfigs(final Map<String, String> properties) {
    final String name = properties.get("name");
    return Optional.of(new Connector(
        name,
        DataSourceType.KTABLE,
        extractKeyNameFromSmt(properties).orElse(null)
    ));
  }
@Test
  public void shouldCreateJdbcConnectorWithValidConfigsAndSMT() {
    // Given:
    final Map<String, String> config = ImmutableMap.of(
        Connectors.CONNECTOR_CLASS, JdbcSource.JDBC_SOURCE_CLASS,
        "name", "foo",
        "transforms", "foobar,createKey",
        "transforms.createKey.type", "org.apache.kafka.connect.transforms.ExtractField$Key",
        "transforms.createKey.field", "key"
    );

    // When:
    final Optional<Connector> maybeConnector = jdbcSource.fromConfigs(config);

    // Then:
    final Connector expected = new Connector(
        "foo",
        DataSourceType.KTABLE,
        "key");
    assertThat(maybeConnector, OptionalMatchers.of(is(expected)));
  }