Kafka Connect protobuf converter not found

Viewed 82

I have a kafka s3-sink connector which I'm producing messages to. However when I send messages to the topic, I'm noticied the logs of my sink connector are throwing me this error:

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

Here's my sink:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: test-s3-sink
  labels:
    # The strimzi.io/cluster label identifies the KafkaConnect instance
    # in which to create this connector. That KafkaConnect instance
    # must have the strimzi.io/use-connector-resources annotation
    # set to true.
    strimzi.io/cluster: test-cluster
spec:
  class: io.confluent.connect.s3.S3SinkConnector
  tasksMax: 6
  config:
    name: test-s3-sink
    topics: testing-stack-overflow-topic
    flush.size: 100
    s3.bucket.name: test-bucket
    s3.region: us-east-1
    storage.class: io.confluent.connect.s3.storage.S3Storage
    format.class: io.confluent.connect.s3.format.avro.AvroFormat
    store.kafka.keys: true
    keys.format.class: io.confluent.connect.s3.format.avro.AvroFormat
    key.converter: "io.confluent.connect.protobuf.ProtobufConverter"
    key.converter.schema.registry.url: "http://localhost:8081"
    key.converter.schemas.enable: true
    value.converter: "io.confluent.connect.protobuf.ProtobufConverter"
    value.converter.schema.registry.url: "http://localhost:8081"
    value.converter.schemas.enable: true
    rotate.interval.ms: 3600000
    timezone: UTC
    behavior.on.null.values: ignore
    partitioner.class: "io.confluent.connect.storage.partitioner.HourlyPartitioner"
    locale: "en-US"

Here's my producer:

package kafka;

import com.github.javafaker.Faker;
import java.util.Random;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer;
import com.example.CardData;
import java.util.Properties;

public class SendKafkaProto {
  public static void main(String[] args) {
    // Setup Producer Properties
    String bootstrapServers = "127.0.0.1:9092";
    var properties = new Properties();
    properties.setProperty("bootstrap.servers", bootstrapServers);
    properties.setProperty("schema.registry.url", "http://localhost:8081");
    properties.setProperty("key.serializer", StringSerializer.class.getName());
    properties.setProperty("value.serializer", KafkaProtobufSerializer.class.getName());

    KafkaProducer<String, CardData.CreditCard> producer = new KafkaProducer<>(properties);
    // Specify Topic Name
    var topic = "protos_topic_cards";

    // Loop to Produce Fake Data
    for (int i = 0; i < 15; i++) {
      // creating Random object
      Random rd = new Random();
      Faker faker = new Faker();
      String name = faker.name().fullName();
      String countryCode = faker.address().countryCode();
      String cardNumber = faker.business().creditCardNumber();
      Integer typeValue = rd.nextInt(3);
      String currencyCode = faker.country().currencyCode();

      // Serializing to Protobuf based on CreditCard.proto Schema
      var cardData = CardData.CreditCard.newBuilder()
          .setName(name)
          .setCountry(countryCode)
          .setCurrency(currencyCode)
          .setTypeValue(typeValue)
          .setBlocked(false)
          .setCardNumber(cardNumber)
          .build();

      var record = new ProducerRecord<String, CardData.CreditCard>(topic, "Credit Card", cardData);
      // Send to Producer
      producer.send(record);
    }
    producer.flush();
    producer.close();
    // Log success message
    System.out.println("Sent Data Successfully");
  }
}

What could amiss here? Do I need to add the Protobuf converter to wherever my kafka image is mounted? Do I need to use a string converter for the key serializer? There aren't many solutions here online. Thanks!

2 Answers

The Strimzi containers don't contain any converters outside the ones included by Apache Kafka.

You need to install it, potentially requiring your own Docker image - https://www.confluent.io/hub/confluentinc/kafka-connect-protobuf-converter

Also, you cannot use localhost as the Registry URL from the Strimzi config; it needs to be an external address outside the Connect container environment, e.g. http://schema-registry.svc.cluster.local:8081 if you had a schema-registry service in the same namespace.

And your producer uses String keys, so key.converter shouldn't be Protobuf ; Strings don't have schemas, so there is nothing to enable. And Protobuf always has a schema, so having schemas.enable doesn't do anything.

If you want to store Avro in S3, using an Avro Producer might make more sense. Again, install the necessary Converter.

Looks like io.confluent.connect.protobuf.ProtobufConverter is not present in your application classpath. Take a look across the JARs that you are executing and whether any of them delivers it and make sure it is present in your path.

Related