Reactive Extensions wrapper around Confluent.Kafka for .NET

Question

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())) 
                        .Build();

                    Task.Run(
                        () =>
                        {
                            try
                            {
                                var (topic, partitionOrNull, offsetOrNull) = target;

                                if (partitionOrNull == null && offsetOrNull == null) consumer.Subscribe(topic);
                                else
                                {
                                    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)
                            {
                                observer.OnError(
                                    // Confluent.Kafka stuff is not allowed to leak out
                                    kafkaException.Error?.AsKafkaException() ?? new KafkaException());
                            }
                            catch (Exception exception)
                            {
                                observer.OnError(exception);
                            }
                        },
                        cancellationToken);

                    // 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);
                });
        }
    }
}