Just an Rx wrapper around Kafka to represent topic consumption as IObservable<T>:

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Confluent.Kafka;
using PM.TT.Live.Kafka.Rx.ConfluentKafka;
using Reactive = System.Reactive;

namespace PM.TT.Live.Kafka.Rx
    public static class Consumer
        /// <summary>
        /// Creates an <see cref="IObservable{T}"/> for a given Kafka topic.
        /// Note that autocommiting must be enabled; there is no support for manual commits.
        /// Will use GUID as 'group.id' if not provided.
        /// </summary>
        /// <typeparam name="TKey">Kafka topic record's key.</typeparam>
        /// <typeparam name="TValue">Kafka topic record's value.</typeparam>
        /// <param name="target">A triple (topic name, partition id, offset) that represents a consumption target.</param>
        /// <param name="keyDeserializer">How to deserialize key.</param>
        /// <param name="valueDeserializer">How to deserialize value.</param>
        /// <param name="kafkaConsumerConfig">
        /// Kafka consumer's config.
        /// Stuff like 'AutoOffsetReset' or 'GroupId' goes here.
        /// See https://kafka.apache.org/documentation/#consumerconfigs for details.
        /// </param>
        /// <returns><see cref="IObservable{T}"/> of a triplet (record offset, record key, record value).</returns>
        public static IObservable<(long offset, TKey key, TValue value)> OfTopic<TKey, TValue>(
            (string topic, int? partition, long? offset) target,
            Func<byte[], TKey> keyDeserializer,
            Func<byte[], TValue> valueDeserializer,
            IReadOnlyDictionary<string, string> kafkaConsumerConfig = default)
            var finalKafkaConsumerConfig = new Dictionary<string, string>();

            if (kafkaConsumerConfig != null)
                foreach (var pair in kafkaConsumerConfig)
                    finalKafkaConsumerConfig.Add(pair.Key, pair.Value);

            if (target.partition == null && target.offset == null && !finalKafkaConsumerConfig.ContainsKey("group.id")) 
                // group.id is a must when doing .subscribe(); 
                // it is not need when .assign()'ing manually to a specific partition
                finalKafkaConsumerConfig.Add("group.id", Guid.NewGuid().ToString());

            return Reactive.Linq.Observable.Create<(long, TKey, TValue)>(
                (observer, cancellationToken) =>
                    var consumer = new ConsumerBuilder<TKey, TValue>(finalKafkaConsumerConfig)
                        .SetKeyDeserializer(new FuncToDeserializer<TKey>(keyDeserializer))
                        .SetValueDeserializer(new FuncToDeserializer<TValue>(valueDeserializer))
                        .SetErrorHandler((_, error) => observer.OnError(error.AsKafkaException())) 

                        () =>
                                var (topic, partitionOrNull, offsetOrNull) = target;

                                if (partitionOrNull == null && offsetOrNull == null) consumer.Subscribe(topic);
                                    var partition = partitionOrNull ?? 0;
                                    if (offsetOrNull == null) consumer.Assign(new TopicPartition(topic, partition));
                                    else consumer.Assign(new TopicPartitionOffset(topic, partition, offsetOrNull.Value));

                                // keeps consuming until cancellation is requested, i.e. until unsubscribed;
                                // note that there is no need to .OnComplete(): Rx framework will handle it itself once cancelled
                                while (!cancellationToken.IsCancellationRequested)
                                    var consumeResult = consumer.Consume(cancellationToken);

                                    if (consumeResult.IsPartitionEOF) continue;

                                    observer.OnNext((consumeResult.Offset.Value, consumeResult.Key, consumeResult.Value));
                            catch (Confluent.Kafka.KafkaException kafkaException)
                                    // Confluent.Kafka stuff is not allowed to leak out
                                    kafkaException.Error?.AsKafkaException() ?? new KafkaException());
                            catch (Exception exception)

                    // it is important to enforce IDisposable as a return value, not just Task (which is deduced automatically);
                    // otherwise Rx framework will dispose consumer as soon as it gets created
                    return Task.FromResult<IDisposable>(consumer);
  • \$\begingroup\$ Hey Sereja. Did you ever get an opinion on this? I am trying to solve the exact same problem and wondering what you ended up doing. \$\endgroup\$
    – stripathi
    Mar 18, 2021 at 7:00
  • \$\begingroup\$ @stripathi a bit simplified version of the code above works well for consumption and successfully runs in production. \$\endgroup\$
    – Zazeil
    Mar 18, 2021 at 7:07


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.