feat: add support for Kafka Schema Registry serialization and deserialization

Author: JandersonConstantino

src/.idea/.idea.Take.Elephant/.idea/indexLayout.xml

@@ -23,35 +23,44 @@ Docker Compose
 
 Create one file named `docker-compose.yml` with this content:
 ```
-version: "3"
 services:
-  zookeeper:
-    image: "confluentinc/cp-zookeeper"
+  broker:
+    image: confluentinc/cp-kafka:8.0.0
+    hostname: broker
+    container_name: broker
     ports:
-      - 2181:2181
+      - "9092:9092"
+      - "9101:9101"
     environment:
-      - ZOOKEEPER_CLIENT_PORT=2181
-    volumes:
-      - zookeeperData:/var/lib/zookeeper/data
-      - zookeeperLogs:/var/lib/zookeeper/log
-
-  kafka:
-    image: "confluentinc/cp-kafka"
+      KAFKA_NODE_ID: 1
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
+      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092'
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_JMX_PORT: 9101
+      KAFKA_JMX_HOSTNAME: localhost
+      KAFKA_PROCESS_ROLES: 'broker,controller'
+      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@broker:29093'
+      KAFKA_LISTENERS: 'PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092'
+      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
+      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
+      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
+      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
+
+  schema-registry:
+    image: confluentinc/cp-schema-registry:8.0.0
+    hostname: schema-registry
+    container_name: schema-registry
+    depends_on:
+      - broker
     ports:
-      - 9092:9092
+      - "8081:8081"
     environment:
-      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
-      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
-      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
-    volumes:
-      - kafka:/var/lib/kafka/data
-    depends_on:
-      - zookeeper
-volumes:
-  kafka:
-  zookeeperData:
-  zookeeperLogs:
-
+      SCHEMA_REGISTRY_HOST_NAME: schema-registry
+      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
+      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
 ```
 
 Basically this file is starting two services on your machine, the `Zookeeper` and the `kafka` itself. Inside the folder that you created the docker compose file, run the command `docker-compose up -d`, the `-d` is for second plan run. After the command run sucessfully you should run `docker-compose ps` to check if the services are running on your pc. You should see:
@@ -86,4 +95,4 @@ kafka_1      | [2019-11-25 16:39:34,220] INFO [KafkaServer id=1] started (kafka.
 ```
 #### Check logs on kafka to see events of consuming and producing
 
-`docker-compose logs --f`
+`docker-compose logs --f`

src/.idea/.idea.Take.Elephant/.idea/modules.xml

@@ -0,0 +1,260 @@
+using System;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Confluent.Kafka;
+using Confluent.SchemaRegistry;
+
+namespace Take.Elephant.Kafka.SchemaRegistry
+{
+    /// <summary>
+    /// A Kafka event stream consumer that uses Schema Registry for deserialization.
+    /// </summary>
+    /// <typeparam name="TKey">The type of the message key.</typeparam>
+    /// <typeparam name="TEvent">The type of the event.</typeparam>
+    public class KafkaSchemaRegistryEventStreamConsumer<TKey, TEvent> : IEventStreamConsumer<TKey, TEvent>, IOpenable, ICloseable, IDisposable
+        where TEvent : class
+    {
+        private readonly IConsumer<TKey, TEvent> _consumer;
+        private readonly ISchemaRegistryClient _schemaRegistryClient;
+        private readonly bool _ownsSchemaRegistryClient;
+        private readonly SemaphoreSlim _consumerStartSemaphore;
+        private readonly CancellationTokenSource _cts;
+        private readonly Channel<Message<TKey, TEvent>> _channel;
+        private Task _consumerTask;
+        private bool _closed;
+
+        /// <summary>
+        /// Creates a new instance of <see cref="KafkaSchemaRegistryEventStreamConsumer{TKey, TEvent}"/>.
+        /// </summary>
+        /// <param name="bootstrapServers">The Kafka bootstrap servers.</param>
+        /// <param name="topic">The topic name.</param>
+        /// <param name="groupId">The consumer group ID.</param>
+        /// <param name="schemaRegistryOptions">The Schema Registry options.</param>
+        public KafkaSchemaRegistryEventStreamConsumer(
+            string bootstrapServers,
+            string topic,
+            string groupId,
+            SchemaRegistryOptions schemaRegistryOptions)
+            : this(
+                new ConsumerConfig { BootstrapServers = bootstrapServers, GroupId = groupId },
+                topic,
+                schemaRegistryOptions)
+        {
+        }
+
+        /// <summary>
+        /// Creates a new instance of <see cref="KafkaSchemaRegistryEventStreamConsumer{TKey, TEvent}"/>.
+        /// </summary>
+        /// <param name="consumerConfig">The consumer configuration.</param>
+        /// <param name="topic">The topic name.</param>
+        /// <param name="schemaRegistryOptions">The Schema Registry options.</param>
+        public KafkaSchemaRegistryEventStreamConsumer(
+            ConsumerConfig consumerConfig,
+            string topic,
+            SchemaRegistryOptions schemaRegistryOptions)
+            : this(
+                consumerConfig,
+                topic,
+                new CachedSchemaRegistryClient(schemaRegistryOptions.SchemaRegistryConfig),
+                schemaRegistryOptions,
+                ownsSchemaRegistryClient: true)
+        {
+        }
+
+        /// <summary>
+        /// Creates a new instance of <see cref="KafkaSchemaRegistryEventStreamConsumer{TKey, TEvent}"/>.
+        /// </summary>
+        /// <param name="consumerConfig">The consumer configuration.</param>
+        /// <param name="topic">The topic name.</param>
+        /// <param name="schemaRegistryClient">The Schema Registry client.</param>
+        /// <param name="schemaRegistryOptions">The Schema Registry options.</param>
+        public KafkaSchemaRegistryEventStreamConsumer(
+            ConsumerConfig consumerConfig,
+            string topic,
+            ISchemaRegistryClient schemaRegistryClient,
+            SchemaRegistryOptions schemaRegistryOptions)
+            : this(consumerConfig, topic, schemaRegistryClient, schemaRegistryOptions, ownsSchemaRegistryClient: false)
+        {
+        }
+
+        private KafkaSchemaRegistryEventStreamConsumer(
+            ConsumerConfig consumerConfig,
+            string topic,
+            ISchemaRegistryClient schemaRegistryClient,
+            SchemaRegistryOptions schemaRegistryOptions,
+            bool ownsSchemaRegistryClient)
+        {
+            if (string.IsNullOrWhiteSpace(topic))
+                throw new ArgumentException("Value cannot be null or whitespace.", nameof(topic));
+
+            Topic = topic;
+            _schemaRegistryClient = schemaRegistryClient ?? throw new ArgumentNullException(nameof(schemaRegistryClient));
+            _ownsSchemaRegistryClient = ownsSchemaRegistryClient;
+
+            var deserializer = SchemaRegistrySerializerFactory.CreateDeserializer<TEvent>(schemaRegistryClient, schemaRegistryOptions);
+            _consumer = new ConsumerBuilder<TKey, TEvent>(consumerConfig)
+                .SetValueDeserializer(deserializer)
+                .Build();
+
+            _consumerStartSemaphore = new SemaphoreSlim(1, 1);
+            _cts = new CancellationTokenSource();
+            _channel = Channel.CreateBounded<Message<TKey, TEvent>>(1);
+        }
+
+        /// <summary>
+        /// Creates a new instance of <see cref="KafkaSchemaRegistryEventStreamConsumer{TKey, TEvent}"/> using a pre-built consumer.
+        /// </summary>
+        /// <param name="consumer">The Kafka consumer.</param>
+        /// <param name="topic">The topic name.</param>
+        /// <param name="schemaRegistryClient">The Schema Registry client (optional, for lifecycle management).</param>
+        public KafkaSchemaRegistryEventStreamConsumer(
+            IConsumer<TKey, TEvent> consumer,
+            string topic,
+            ISchemaRegistryClient schemaRegistryClient = null)
+        {
+            if (string.IsNullOrWhiteSpace(topic))
+                throw new ArgumentException("Value cannot be null or whitespace.", nameof(topic));
+
+            _consumer = consumer ?? throw new ArgumentNullException(nameof(consumer));
+            Topic = topic;
+            _schemaRegistryClient = schemaRegistryClient;
+            _ownsSchemaRegistryClient = false;
+            _consumerStartSemaphore = new SemaphoreSlim(1, 1);
+            _cts = new CancellationTokenSource();
+            _channel = Channel.CreateBounded<Message<TKey, TEvent>>(1);
+        }
+
+        /// <summary>
+        /// Gets the topic name.
+        /// </summary>
+        public string Topic { get; }
+
+        /// <summary>
+        /// Occurs when the consumer fails.
+        /// </summary>
+        public event EventHandler<ExceptionEventArgs> ConsumerFailed;
+
+        /// <inheritdoc />
+        public virtual async Task<(TKey key, TEvent item)> ConsumeOrDefaultAsync(CancellationToken cancellationToken)
+        {
+            await StartConsumerTaskIfNotAsync(cancellationToken);
+            if (_channel.Reader.TryRead(out var message))
+            {
+                return (message.Key, message.Value);
+            }
+
+            return default;
+        }
+
+        /// <inheritdoc />
+        public Task OpenAsync(CancellationToken cancellationToken)
+        {
+            return StartConsumerTaskIfNotAsync(cancellationToken);
+        }
+
+        /// <inheritdoc />
+        public virtual async Task CloseAsync(CancellationToken cancellationToken)
+        {
+            if (!_closed)
+            {
+                _closed = true;
+                await _cts.CancelAsync();
+                if (_consumerTask != null)
+                {
+                    await _consumerTask;
+                }
+                _consumer.Close();
+            }
+        }
+
+        /// <inheritdoc />
+        public void Dispose()
+        {
+            Dispose(true);
+            GC.SuppressFinalize(this);
+        }
+
+        protected virtual void Dispose(bool disposing)
+        {
+            if (disposing)
+            {
+                if (!_closed)
+                {
+                    _consumer.Close();
+                }
+
+                _consumer.Dispose();
+                _cts.Dispose();
+                _consumerStartSemaphore.Dispose();
+
+                if (_ownsSchemaRegistryClient)
+                {
+                    _schemaRegistryClient?.Dispose();
+                }
+            }
+        }
+
+        private async Task StartConsumerTaskIfNotAsync(CancellationToken cancellationToken)
+        {
+            if (_consumerTask != null) return;
+
+            await _consumerStartSemaphore.WaitAsync(cancellationToken);
+            try
+            {
+                if (_consumerTask == null)
+                {
+                    _consumerTask = Task
+                        .Factory
+                        .StartNew(
+                            () => ConsumeAsync(_cts.Token),
+                            TaskCreationOptions.LongRunning)
+                        .Unwrap();
+                }
+            }
+            finally
+            {
+                _consumerStartSemaphore.Release();
+            }
+        }
+
+        private async Task ConsumeAsync(CancellationToken cancellationToken)
+        {
+            if (_consumer.Subscription.All(s => s != Topic))
+            {
+                _consumer.Subscribe(Topic);
+            }
+
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                try
+                {
+                    var result = _consumer.Consume(cancellationToken);
+                    await _channel.Writer.WriteAsync(result.Message, cancellationToken);
+                }
+                catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+                {
+                    break;
+                }
+                catch (Exception ex)
+                {
+                    var handler = ConsumerFailed;
+                    if (handler != null)
+                    {
+                        handler.Invoke(this, new ExceptionEventArgs(ex));
+                    }
+                    else
+                    {
+                        Trace.TraceError("An unhandled exception occurred on KafkaSchemaRegistryEventStreamConsumer: {0}", ex);
+                    }
+                }
+            }
+
+            _consumer.Unsubscribe();
+            _channel.Writer.Complete();
+        }
+    }
+}
+