Kafka - Avro

This sample implements a producer and consumer which take advantage of the schema registry and serializes the messages as Avro.

Common

The message being exchanged is defined in a common project.

// ------------------------------------------------------------------------------
// <auto-generated>
//    Generated by avrogen, version 1.7.7.5
//    Changes to this file may cause incorrect behavior and will be lost if code
//    is regenerated
// </auto-generated>
// ------------------------------------------------------------------------------

using Silverback.Messaging.Messages;

namespace Silverback.Examples.Messages
{
    using System;
    using System.Collections.Generic;
    using System.Text;
    using global::Avro;
    using global::Avro.Specific;
    
    public partial class AvroMessage : ISpecificRecord
    {
        public static Schema _SCHEMA = Schema.Parse("{\"type\":\"record\",\"name\":\"AvroMessage\",\"namespace\":\"Silverback.Examples.Messages\"," +
                                                    "\"fields\":[{\"name\":\"number\",\"type\":\"string\"}]}");
        private string _number;
        public virtual Schema Schema
        {
            get
            {
                return AvroMessage._SCHEMA;
            }
        }
        public string number
        {
            get
            {
                return this._number;
            }
            set
            {
                this._number = value;
            }
        }
        public virtual object Get(int fieldPos)
        {
            switch (fieldPos)
            {
                case 0: return this.number;
                default: throw new AvroRuntimeException("Bad index " + fieldPos + " in Get()");
            }
        }
        public virtual void Put(int fieldPos, object fieldValue)
        {
            switch (fieldPos)
            {
                case 0:
                    this.number = (System.String) fieldValue;
                    break;
                default: throw new AvroRuntimeException("Bad index " + fieldPos + " in Put()");
            }
        }
    }
}

Full source code: https://github.com/BEagle1984/silverback/tree/master/samples/Kafka/Avro.Common

Producer

The producer uses a hosted service to publish some messages in the background.

using Microsoft.Extensions.DependencyInjection;
using Silverback.Configuration;
using Silverback.Messaging.Configuration;

namespace Silverback.Samples.Kafka.Avro.Producer;

public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        // Enable Silverback
        services.AddSilverback()

            // Use Apache Kafka as message broker and the Confluent schema registry
            .WithConnectionToMessageBroker(options => options
                .AddKafka()
                .AddConfluentSchemaRegistry())

            // Delegate the broker clients configuration to a separate class
            .AddBrokerClientsConfigurator<BrokerClientsConfigurator>();

        // Add the hosted service that produces the random sample messages
        services.AddHostedService<ProducerBackgroundService>();
    }

    public void Configure()
    {
    }
}

using Silverback.Examples.Messages;
using Silverback.Messaging.Configuration;

namespace Silverback.Samples.Kafka.Avro.Producer;

public class BrokerClientsConfigurator : IBrokerClientsConfigurator
{
    public void Configure(BrokerClientsConfigurationBuilder builder)
    {
        builder
            .AddKafkaClients(
                clients => clients

                    // The bootstrap server address is needed to connect
                    .WithBootstrapServers("PLAINTEXT://localhost:19092")

                    // Add a producer
                    .AddProducer(
                        producer => producer

                            // Produce the AvroMessage to the samples-avro topic
                            .Produce<AvroMessage>(
                                endpoint => endpoint
                                    .ProduceTo("samples-avro")
                                    .SetKafkaKey(message => message?.number.ToString())

                                    // Configure Avro serialization
                                    .SerializeAsAvro(
                                        avro => avro
                                            .ConnectToSchemaRegistry("localhost:8081")
                                            .Configure(
                                                serializer =>
                                                {
                                                    serializer.AutoRegisterSchemas = false;
                                                    serializer.UseLatestVersion = true;
                                                })))));
    }
}

using System;
using System.Globalization;
using System.Threading;
using System.Threading.Tasks;
using Confluent.SchemaRegistry;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Silverback.Examples.Messages;
using Silverback.Messaging.Publishing;
using Silverback.Messaging.Serialization;

namespace Silverback.Samples.Kafka.Avro.Producer;

public class ProducerBackgroundService : BackgroundService
{
    private readonly IConfluentSchemaRegistryClientFactory _schemaRegistryClientFactory;

    private readonly IPublisher _publisher;

    private readonly ILogger<ProducerBackgroundService> _logger;

    public ProducerBackgroundService(
        IConfluentSchemaRegistryClientFactory schemaRegistryClientFactory,
        IPublisher publisher,
        ILogger<ProducerBackgroundService> logger)
    {
        _schemaRegistryClientFactory = schemaRegistryClientFactory;
        _publisher = publisher;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // Push the Avro schema to the schema registry
        ISchemaRegistryClient schemaRegistry = _schemaRegistryClientFactory.GetClient(
            schemaRegistryConfigurationBuilder => schemaRegistryConfigurationBuilder
                .WithUrl("localhost:8081"));

        await schemaRegistry.RegisterSchemaAsync(
            "samples-avro-value",
            new Schema(AvroMessage._SCHEMA.ToString(), null, SchemaType.Avro, null, null));

        int number = 0;

        while (!stoppingToken.IsCancellationRequested)
        {
            await ProduceMessageAsync(++number);

            await Task.Delay(100, stoppingToken);
        }
    }

    private async Task ProduceMessageAsync(int number)
    {
        try
        {
            await _publisher.PublishAsync(
                new AvroMessage
                {
                    number = number.ToString(CultureInfo.InvariantCulture)
                });

            _logger.LogInformation("Produced {Number}", number);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to produce {Number}", number);
        }
    }
}

Full source code: https://github.com/BEagle1984/silverback/tree/master/samples/Kafka/Avro.Producer

Consumer

The consumer processes the messages and outputs their value to the standard output.

using Microsoft.Extensions.DependencyInjection;
using Silverback.Configuration;
using Silverback.Messaging.Configuration;

namespace Silverback.Samples.Kafka.Avro.Consumer;

public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        // Enable Silverback
        services.AddSilverback()

            // Use Apache Kafka as message broker and the Confluent schema registry
            .WithConnectionToMessageBroker(options => options
                .AddKafka()
                .AddConfluentSchemaRegistry())

            // Delegate the broker clients configuration to a separate class
            .AddBrokerClientsConfigurator<BrokerClientsConfigurator>()

            // Register the subscribers
            .AddSingletonSubscriber<AvroMessageSubscriber>();
    }

    public void Configure()
    {
    }
}

using Silverback.Examples.Messages;
using Silverback.Messaging.Configuration;

namespace Silverback.Samples.Kafka.Avro.Consumer;

public class BrokerClientsConfigurator : IBrokerClientsConfigurator
{
    public void Configure(BrokerClientsConfigurationBuilder builder)
    {
        builder
            .AddKafkaClients(
                clients => clients

                    // The bootstrap server address is needed to connect
                    .WithBootstrapServers("PLAINTEXT://localhost:19092")

                    // Add a consumer
                    .AddConsumer(
                        consumer => consumer

                            // Set the consumer group id
                            .WithGroupId("sample-consumer")

                            // AutoOffsetReset.Earliest means that the consumer
                            // must start consuming from the beginning of the topic,
                            // if no offset was stored for this consumer group
                            .AutoResetOffsetToEarliest()

                            // Consume the AvroMessage from the samples-avro topic
                            .Consume<AvroMessage>(
                                endpoint => endpoint
                                    .ConsumeFrom("samples-avro")

                                    // Configure Avro deserialization
                                    .DeserializeAvro(
                                        avro => avro
                                            .ConnectToSchemaRegistry("localhost:8081")))));
    }
}

using Microsoft.Extensions.Logging;
using Silverback.Examples.Messages;

namespace Silverback.Samples.Kafka.Avro.Consumer;

public class AvroMessageSubscriber
{
    private readonly ILogger<AvroMessageSubscriber> _logger;

    public AvroMessageSubscriber(ILogger<AvroMessageSubscriber> logger)
    {
        _logger = logger;
    }

    public void OnMessageReceived(AvroMessage message) =>
        _logger.LogInformation("Received {MessageNumber}", message.number);
}

Full source code: https://github.com/BEagle1984/silverback/tree/master/samples/Kafka/Avro.Consumer

Table of Contents

Kafka - Avro

Common

Producer

Consumer