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!