Kafka-connect-jdbc: problem applying flatten transformation

[aschofie]

I am attempting to flatten a topic before sending it along to my postgres db, using something like the connector below. I am using the confluent 4.1.1 kafka connect docker image, the only change being I copied a custom connector jar into /usr/share/java and am running it under a different accoount.

version (kafka connect) "1.1.1-cp1"

commit "0a5db4d59ee15a47"

{
    "name": "problematic_postgres_sink",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",

        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "key.converter.schema.registry.url": "http://kafkaschemaregistry.service.consul:8081",

        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "value.converter.schema.registry.url": "http://kafkaschemaregistry.service.consul:8081",

        "connection.url": "jdbc:postgresql://123.123.123.123:5432/mypostgresdb",
        "connection.user": "abc",
        "connection.password": "xyz",

        "insert.mode": "upsert",
        "auto.create": true,
        "auto.evolve": true,

        "topics": "mytopic",
        "pk.mode": "kafka",

        "transforms": "Flatten",
        "transforms.Flatten.type": "org.apache.kafka.connect.transforms.Flatten$Value",
        "transforms.Flatten.delimiter": "_"
    }
}

I get a 400 error code:

Connector configuration is invalid and contains the following 1
error(s): Invalid value class
org.apache.kafka.connect.transforms.Flatten for configuration
transforms.Flatten.type: Error getting config definition from
Transformation: null

[AnkushKhanna]

Use "org.apache.kafka.connect.transforms.Flatten\$Value" instead.