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\$
    – Zazaeil
    Mar 18, 2021 at 7:07


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Browse other questions tagged or ask your own question.