How to set Protobuf Serialization With Kafka Using Confluent Schema Registry Using C++

[alchemistsrivastava]

Here are ways to integrate protobuf schema using confluent schema registry with Java and Python. How can we support the protobuf schema using C++?

USING JAVA
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaProtobufSerializer.class);
properties.put(KafkaProtobufSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
Producer<String, SimpleMessage> producer = new KafkaProducer<>(properties);

USING PYTHON
from confluent_kafka import SerializingProducer
from confluent_kafka.serialization import StringSerializer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.protobuf import ProtobufSerializer
topic = 'TEST_PROTO'
schema_registry_client = SchemaRegistryClient({'url': 'http://localhost:8081'})
protobuf_serializer = ProtobufSerializer(test_pb2.Meal, schema_registry_client)
producer_conf = {'bootstrap.servers': 'localhost:9092', 'key.serializer': StringSerializer('utf_8'), 'value.serializer': protobuf_serializer}
producer = SerializingProducer(producer_conf)

QUESTION: How to get the Kafka producer instance using schema registry in C++? OR how to publish protobuf messages using Kafka producer in C++

Thanks so much for your attention

[alchemistsrivastava]

ADDING these properties did NOT work if (rd_kafka_conf_set(conf, "key.serializer", "org.apache.kafka.common.serialization.StringSerializer",
errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
fprintf(stderr, "%% %s\n", errstr);
exit(1);
}
if (rd_kafka_conf_set(conf, "value.serializer", "io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer",
errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
fprintf(stderr, "%% %s\n", errstr);
exit(1);
}
if (rd_kafka_conf_set(conf, "schema.registry.url", "localhost:9092",
errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
fprintf(stderr, "%% %s\n", errstr);
exit(1);
}

[alchemistsrivastava]

Can anyone please suggest how to set configuration properties to enable protobuf with Kafka on C++

[RPG-18]

You will have to implement it yourself. You can see sources https://github.com/edenhill/kcat and https://github.com/confluentinc/libserdes.

[le-cardinal]

btw, it will be an nice feature to be added in the next releases of librdkafka