Is there a way to use confluent Kafka Dotnet JSON serializer WITHOUT schema registry,

Viewed 2553
2 Answers

The link you've provided is for JSON Schema, not plain JSON.

JSON is a plaintext format. Use the Utf8Serializer and send strings after converting any model class or dictionary into a JSON string.

However, this doesn't guarantee (on the server-side) that your messages adhere to any agreed upon format (i.e. a required set of fields is defined), so that's where you'd want introduce a schema and use a registry

Try this:

Model:

public class KafkaMessage : ISerializer<KafkaMessage>, IDeserializer<KafkaMessage>
{
    public MessageType MessageType { get; set; }
    public string MessageData { get; set; }

    public byte[] Serialize(KafkaMessage data, SerializationContext context)
    {
        using (var ms = new MemoryStream())
        {
            string jsonString = JsonSerializer.Serialize(data);
            var writer = new StreamWriter(ms);

            writer.Write(jsonString);
            writer.Flush();
            ms.Position = 0;

            return ms.ToArray();
        }
    }

    KafkaMessage IDeserializer<KafkaMessage>.Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
    {
        return JsonSerializer.Deserialize<KafkaMessage>(data.ToArray());
    }
}

Producer:

using (var producer = new ProducerBuilder<Null, KafkaMessage>(_config)
    .SetValueSerializer(new KafkaMessage())
    .Build())
{
    KafkaMessage km = new KafkaMessage { MessageType = messageType, MessageData = messageData };
    var deliveryResult = await producer
        .ProduceAsync(_topic, new Message<Null, KafkaMessage> { Value = km })
        .ContinueWith(task => !task.IsFaulted);

    result = deliveryResult;
}

Consumer:

using (var consumer = new ConsumerBuilder<Ignore, KafkaMessage>(_config)
    .SetValueDeserializer(new KafkaMessage())
    .SetLogHandler((_, logHandler) => {  })
    .SetErrorHandler((_, errorHandler) => { })
    .Build())
{ 
    consumer.Subscribe(_topic);
    try
    {
        while (!token.IsCancellationRequested)
        {
            var result = consumer.Consume(token);
            var messageData = result.Message.Value;
            Console.log($"Output: { messageData.MessageData }")
        }
    }
    catch (OperationCanceledException)
    {
        /* Close connection */
        consumer.Close();
    }
}
Related