Kafka Connect Protobuf Configuration

user_7694272 0
Viewed 2243

I'm trying to create a kafka sink connector that uses a protobuf value converter. I've got a version of this configuration working with JSON, however I now need to change it to use protobuf messages.

I'm trying to create a connector with the following request:

curl -X POST localhost:8083/connectors -H "Content-Type: application/json" -d '
{
    "name": "jdbc-sink-connector",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
      "topics": "TEST_PROTO",
      "connection.url": "${DB_URL}",
      "value.converter": "io.confluent.connect.protobuf.ProtobufConverter",
      "key.converter": "io.confluent.connect.protobuf.ProtobufConverter",
      "auto.create": true,
      "auto.evolve": true,
      "type": "sink",
      "connection.user": "{DB_USER}",
      "connection.password": "${DB_PASS}"
    }
}

This gives the following 400 error message:

Invalid value io.confluent.connect.protobuf.ProtobufConverter for configuration value.converter: Class io.confluent.connect.protobuf.ProtobufConverter could not be found

I'm not quite understanding why I'm not able to include this here. From what I can see the documentation suggests this is an appropriate value: https://docs.confluent.io/current/connect/userguide.html

Please can anyone help?

apache-kafka-connect apache-kafka grpc protocol-buffers
2 Answers

I guess that in this case, you are missing these configurations:

  • value.converter.schema.registry.url
  • key.converter.schema.registry.url
  • key.converter.schemas.enable
  • value.converter.schemas.enable

I addition to these also try using the latest jars of jdbc and latest version for confluent platform. If this doesn't works please let me know.

user_147442880

Adding more to the above answer for clarity, you need to mention the values for the below keys while configuring kafka-connect.

value.converter = "io.confluent.connect.protobuf.ProtobufConverter"
key.converter = "io.confluent.connect.protobuf.ProtobufConverter"
value.converter.schema.registry.url = URL (You should have schema registry service installed and all the producers registering the schema to the service registry before writing to the broker)
key.converter.schema.registry.url = Can be the same URL as above
key.converter.schemas.enable = true (if using Protobuf)
value.converter.schemas.enable = true (if using Protobuf)

To verify if the converter is loaded successfully, you can check the INFO logs.

user_70646270
Related