CancelToken seems to not work properly on consumer

[tsebastiani]

Description

Hi,
I'm trying to implement a simple graceful shutdown for kafka consumer. The expected behaviour (from the api description) should be that the consume method waits until a new message is fetched or the cancel method is called on the CancellationTokenSource, but this is not actually happening.
My first try was to run the Polling from inside a Task, but it wasn't working at all so I tried with the simplest example possible, but it's still not working.

Confluent nuget version 1.3.0-PRE1 , 1.2.1, 1.2.0
Kafka version 2.12-2.3.0
DotnetCore 3.0
MacOS Catalina

How to reproduce

            var tokenSource = new CancellationTokenSource();
            var token = tokenSource.Token;
            //Binding the token cancel event to an assembly unload (issued on docker stop) and a ctrl+c
            AssemblyLoadContext.Default.Unloading += (s) => { Logger.Debug("Container closing..."); tokenSource.Cancel(); };
            Console.CancelKeyPress += (s, e) => { tokenSource.Cancel(); Logger.Debug("Container ctrl+c..."); };
            var conf = new ConsumerConfig
            {
                GroupId = "cg1",
                BootstrapServers = "localhost:9092",
                AutoOffsetReset = AutoOffsetReset.Earliest,
            };
            using (var c = new ConsumerBuilder<Ignore, string>(conf).Build())
            {
                c.Subscribe(Env.InputTopic);

                Logger.Debug("Waiting for a message.....");
                var message = c.Consume(token);

                if (token.IsCancellationRequested)
                {
                   //this log is never printed after docker stop or ctrl+c "Container Closing" and "Container ctrl+c..." is printed instead
                    Logger.Debug("Token canceled closing connection");
                }
            }
            //this log is never printed after docker stop ctrl+c "Container Closing" and "Container ctrl+c..." is printed instead
            Logger.Debug("Consumer canceled Exiting....");
            return;

Checklist

Please provide the following information:

• A complete (i.e. we can run it), minimal program demonstrating the problem. No need to supply a project file.
• Confluent.Kafka nuget version.
• Apache Kafka version.
• Client configuration.
• Operating system.
• Provide logs (with "debug" : "..." as necessary in configuration).
• Provide broker log excerpts.
• Critical issue.

[mhowlett]

thanks for reporting and providing a simple example. this should indeed work. i'll will try to replicate before 1.3.

[tsebastiani]

Hi,
(EDITED for a typo in the code wasn't working before) if might be helpful In this way is actually working, but when the Close() is called on the consumer froim the cancel token callback the following error is logged

%3|1571136234.244|ERROR|rdkafka#consumer-1| [thrd:GroupCoordinator]: 1/1 brokers are down

But the brokers are actually working properly.

            List<Task> tasks = new List<Task>();
            var t1 = Task.Run(() =>
            {
                var conf = new ConsumerConfig
                {
                    GroupId = Env.ConsumerGroup,
                    BootstrapServers = Env.BrokerAddress,
                    AutoOffsetReset = AutoOffsetReset.Earliest,
                };

                using (var c = new ConsumerBuilder<Ignore, string>(conf).Build())
                {
                    c.Subscribe(Env.InputTopic);

                    bool taskIsRunning = true;
                    token.Register(() =>
                    {
                        taskIsRunning = false;
                        c.Close();
                        Logger.Debug("KAFKA POLLING LOOP: Cancel callback invoked");
                    });

                    while (taskIsRunning)
                    {
                        Logger.Debug("Polling");
                        var message = c.Consume(new TimeSpan(0, 0, 0, 0, 500));
                    }

                    //this debug is never printed even if the callback ends properly (?)
                    Logger.Debug("KAFKA POLLING LOOP: Consumer closed, exiting");
                    return;
                }
            });
            tasks.Add(t1)
           
            await Tasks.WhenAll(tasks.ToArray());

[PeterWone]

I'm still seeing this in 2023.

Because the consume method fails to exit when cancelled, a call to Dispose on the consumer makes it through an AccessViolationException which is a CST (corrupted state termination) exception and is therefore not caught by exception handlers. Even lamer is how few people have noticed this, which implies most people don't use cancellation tokens and dispose resources properly.

Workaround

This exits properly in response to a cancellation, so dispose doesn't trigger an AccessViolationException that kills the application. If it happens to be sleeping when disposal occurs that's fine because it won't re-enter the loop.

while (!cancellationToken.IsCancellationPending) 
{
  var message = consumer.Consume(0); // timeout of 0 does not wait
  if (null == message)
  {
    Thread.Sleep(5000); //queue is empty, have a nap
  }
  else
  {
    //process the message
  }
}

[gcsizmadia]

It is still an issue in 2024 but thanks to @PeterWone, the workaround helps.