C# Confluent.Kafka SetValueDeserializer object deserialization

Question

In my consumer, I want to deserialize Kafka protobuf message. The key is of string type but message value is a protobuf object. I know I have to create my own custom deserializer for message value but no idea how can I create one. Here is my consumer implementation where I need to replace the marked line:

using Confluent.Kafka;
using System;
using System.Threading;

namespace EventHubsForKafkaSample
{
    class Worker1
    {
        public static void Consumer(string brokerList, string connStr, string consumergroup, string topic, string cacertlocation)
        {
            var config = new ConsumerConfig
            {
                BootstrapServers = brokerList,
                SecurityProtocol = SecurityProtocol.SaslSsl,
                SocketTimeoutMs = 60000,                //this corresponds to the Consumer config `request.timeout.ms`
                SessionTimeoutMs = 30000,
                SaslMechanism = SaslMechanism.Plain,
                SaslUsername = "$ConnectionString",
                SaslPassword = connStr,
                SslCaLocation = cacertlocation,
                GroupId = consumergroup,
                AutoOffsetReset = AutoOffsetReset.Earliest,
                BrokerVersionFallback = "1.0.0",        //Event Hubs for Kafka Ecosystems supports Kafka v1.0+, a fallback to an older API will fail
                //Debug = "security,broker,protocol"    //Uncomment for librdkafka debugging information
            };

            using (var consumer = new ConsumerBuilder<string, ProtobufMessage>(config)
                .SetKeyDeserializer(Deserializers.Utf8)
                .SetValueDeserializer(Deserializers.Utf8) //<<-----
                .Build())
            {
                CancellationTokenSource cts = new CancellationTokenSource();
                Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); };

                consumer.Subscribe(topic);

                Console.WriteLine("Consuming messages from topic: " + topic + ", broker(s): " + brokerList);

                while (true)
                {
                    try
                    {
                        var msg = consumer.Consume(cts.Token);
                        Console.WriteLine($"Received: '{msg.Value}'");
                    }
                    catch (ConsumeException e)
                    {
                        Console.WriteLine($"Consume error: {e.Error.Reason}");
                    }
                    catch (Exception e)
                    {
                        Console.WriteLine($"Error: {e.Message}");
                    }
                }
            }
        }
    }

    public class ProtobufMessage
    {
        public DateTime timestamp { get; set; }
        public int inputId { get; set; }
        public double? value { get; set; }
        public int sourceId { get; set; }
        public string inputGuid { get; set; }
    }
}

Protobuf message format:

syntax = "proto3";

package ileco.chimp.proto;

import "google/protobuf/timestamp.proto";
import "google/protobuf/wrappers.proto";

option java_package = "ileco.chimp.proto";
option java_outer_classname = "FinalValueProtos";

message FinalValue {
  google.protobuf.Timestamp timestamp = 1;
  uint32 inputId = 2;
  google.protobuf.DoubleValue value = 3;
  uint32 sourceId = 4;
  string inputGuid = 5;
}

Answer 1

1. You need to use protoc to generate your C# class from the schema

2. You don't need your own deserializer if you are using the Schema Registry. See example code

    using (var consumer =
        new ConsumerBuilder<string, YourProtoMessage>(consumerConfig)
            .SetValueDeserializer(new ProtobufDeserializer<YourProtoMessage>().AsSyncOverAsync())
   

If you aren't using the Schema Registry, then you will need to define your own deserializer, by implementing IDeserializer, as mentioned in the other answer

Answer 2

You need a class that implements the IDeserializer<T> interface as defined in the Kafka API documentation. Then, your marked line will be something like:

.SetValueDeserializer(new MyCustomDeserializer())

Answer 3

Sample using Protobuf.net custom Serializer

using Confluent.Kafka;
using System.IO;
using SerializationContext = Confluent.Kafka.SerializationContext;

namespace Common;

/// <summary>
/// Kafka protobuf serializer.
/// </summary>
/// <typeparam name="T">Type to serialize. </typeparam>
public class KafkaProtobufSerializer<T> : ISerializer<T> where T : class
{
    /// <inheritdoc/>
    public byte[] Serialize(T data, SerializationContext context)
    {
        using var ms = new MemoryStream();
        ProtoBuf.Serializer.Serialize<T>(ms, data);
        return ms.ToArray();
    }
}
/// <summary>
/// Protobuf deserializer.
/// </summary>
/// <typeparam name="T">Type to deserialize.</typeparam>
public class KafkaProtobufDeserializer<T> : IDeserializer<T> where T : class, new()
{
    /// <inheritdoc/>
    public T Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
    {
        if (isNull || data.IsEmpty)
        {
            return new T();
        }
        
        return ProtoBuf.Serializer.Deserialize<T>(data);
    }
}

Usage Example for the consumer

        var config = new ConsumerConfig
        {
            BootstrapServers = bootstrapServers,
            GroupId = groupId,
            AutoOffsetReset = AutoOffsetReset.Earliest,
            EnableAutoCommit = true,
            EnableAutoOffsetStore = false,
        };

        var consumer = new ConsumerBuilder<TKey, TValue>(config)
            .SetValueDeserializer(new KafkaProtobufDeserializer<TValue>())
            .SetErrorHandler((_, e) => workerLogger.LogError("Consumer Builder:{reason}", e.Reason))
            .Build();