1

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 2

2

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.

3
  • This is the answer, thanks. I'm trying to install the protobuf converter right now. Also, thanks for the schema registry advice. I actually have a external address but just put localhost as a placeholder. Commented Jun 28, 2022 at 20:38
  • After installing, the error persists. Is there some additional configuration required that's documented? Commented Jun 29, 2022 at 18:09
  • How did you install? If you used confluent-hub, then that would modify the plugin.path config in the connect-distributed.properties file. Maybe k8s is overriding this setting? Or if you manually downloaded the converter packages and extracted within the container somewhere, then you need to add that path to that config value. Also see - strimzi.io/blog/2021/03/29/connector-build Commented Jun 29, 2022 at 18:46
0

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.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.