Confluent-kafka-dotnet: Consume method returns an uncheked null ConsumerResult

Created on 28 Aug 2018 · 2Comments · Source: confluentinc/confluent-kafka-dotnet

Description

Consume(CancellationToken cancellationToken) method can return a unchecked null ConsumeResult, throwing a NullReferenceException when the Consumer set up takes more than 100 ms

How to reproduce

Create a new Consumer, no matter what IDeserializers are used, and consume a topic where the set up takes more that 100 ms

How to fix

Checks if result is null in Confluent.Kafka.Consumer class (Note the ? before result)

557 public ConsumeResult<TKey, TValue> Consume(CancellationToken cancellationToken = default(CancellationToken))
...
563 if (result?.Message == null) { continue; }

To avoid an unchecked null return :

369        internal ConsumeResult<TKey, TValue> Consume(int millisecondsTimeout)
370        {
371            var msgPtr = kafkaHandle.ConsumerPoll((IntPtr)millisecondsTimeout);
372            if (msgPtr == IntPtr.Zero)
373            {
374                return null;
375            }

The previous version returns a non null ConsumeResult, then it was right to check the Message property:

return new ConsumeResult<TKey, TValue>
                {
                    TopicPartitionOffset = new TopicPartitionOffset(null, Partition.Any, Offset.Invalid),
                    Message = null
                };