Unable to start multiple Kafka consumers

user_4042929 0
Viewed 392

I have a console app that has multiple BackgroundServices, each reading from the same Kafka topic using the Confluent.Kafka nuget package (v1.6.2). The topic has 3 partitions.

When the app starts, all the background services have their constructors called, however only one of the ExecuteAsync methods is ever called. If I add a Task.Delay() - the number of milliseconds doesn't seem to matter - at the start of each ExecuteAsync, everything works fine and all the background services run.

No exceptions are raised, as far as I can tell.

Does anyone have an idea of what may be happening, or where to look further?

Here's the code:

using Confluent.Kafka;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace KafkaConsumer
{
    class Program
    {
        static void Main(string[] args)
        {
            CreateHostBuilder(args).Build().Run();
        }

        private static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .ConfigureServices((hostContext, services) =>
                {
                    services.AddHostedService<ConsumerA>();
                    services.AddHostedService<ConsumerB>();
                    services.AddHostedService<ConsumerC>();
                });
    }

    public class ConsumerA : BackgroundService
    {
        private readonly ILogger<ConsumerA> _logger;
        private readonly IConsumer<Ignore, string> _consumer;

        public ConsumerA(ILogger<ConsumerA> logger)
        {
            _logger = logger;
            var config = new ConsumerConfig()
            {
                BootstrapServers = @"server:port",
                GroupId = "Group1",
                AutoOffsetReset = AutoOffsetReset.Earliest
            };

            _consumer = new ConsumerBuilder<Ignore, string>(config).Build();
            _logger.LogInformation("ConsumerA constructor");
        }

        protected override async Task ExecuteAsync(CancellationToken cancellationToken)
        {
            // await Task.Delay(10);

            _logger.LogInformation("ConsumerA starting");
            _consumer.Subscribe(new List<string> { "topic" });

            while (!cancellationToken.IsCancellationRequested)
            {
                _ = _consumer.Consume(cancellationToken);
            }
        }
    }

    public class ConsumerB : BackgroundService
    {
        private readonly ILogger<ConsumerB> _logger;
        private readonly IConsumer<Ignore, string> _consumer;

        public ConsumerB(ILogger<ConsumerB> logger)
        {
            _logger = logger;
            var config = new ConsumerConfig()
            {
                BootstrapServers = @"server:port",
                GroupId = "Group1",
                AutoOffsetReset = AutoOffsetReset.Earliest
            };

            _consumer = new ConsumerBuilder<Ignore, string>(config).Build();
            _logger.LogInformation("ConsumerB constructor");
        }

        protected override async Task ExecuteAsync(CancellationToken cancellationToken)
        {
            // await Task.Delay(10);

            _logger.LogInformation("ConsumerB starting");
            _consumer.Subscribe(new List<string> { "topic" });

            while (!cancellationToken.IsCancellationRequested)
            {
                _ = _consumer.Consume(cancellationToken);
            }
        }
    }

    public class ConsumerC : BackgroundService
    {
        private readonly ILogger<ConsumerC> _logger;
        private readonly IConsumer<Ignore, string> _consumer;

        public ConsumerC(ILogger<ConsumerC> logger)
        {
            _logger = logger;
            var config = new ConsumerConfig()
            {
                BootstrapServers = @"server:port",
                GroupId = "Group1",
                AutoOffsetReset = AutoOffsetReset.Earliest
            };

            _consumer = new ConsumerBuilder<Ignore, string>(config).Build();
            _logger.LogInformation("ConsumerC constructor");
        }

        protected override async Task ExecuteAsync(CancellationToken cancellationToken)
        {
            // await Task.Delay(10);
            
            _logger.LogInformation("ConsumerC starting");
            _consumer.Subscribe(new List<string> { "topic" });

            while (!cancellationToken.IsCancellationRequested)
            {
                _ = _consumer.Consume(cancellationToken);
            }
        }
    }
}

And the output:

(with no delays):

info: KafkaConsumer.ConsumerA[0]
      ConsumerA constructor
info: KafkaConsumer.ConsumerB[0]
      ConsumerB constructor
info: KafkaConsumer.ConsumerC[0]
      ConsumerC constructor
info: KafkaConsumer.ConsumerA[0]
      ConsumerA starting

(with delays added):

info: KafkaConsumer.ConsumerA[0]
      ConsumerA constructor
info: KafkaConsumer.ConsumerB[0]
      ConsumerB constructor
info: KafkaConsumer.ConsumerC[0]
      ConsumerC constructor
info: Microsoft.Hosting.Lifetime[0]
      Application started. Press Ctrl+C to shut down.
info: Microsoft.Hosting.Lifetime[0]
      Hosting environment: Production
info: Microsoft.Hosting.Lifetime[0]
      Content root path: c:\users\..\kafkaconsumer\bin\Debug\net5.0
info: KafkaConsumer.ConsumerC[0]
      ConsumerC starting
info: KafkaConsumer.ConsumerA[0]
      ConsumerA starting
info: KafkaConsumer.ConsumerB[0]
      ConsumerB starting
apache-kafka .net-core c#
2 Answers

This is because your execute method is async but you don't use await inside that to inform SynchronizationContext. Write your executeAsync method like this:

protected override async Task ExecuteAsync(CancellationToken cancellationToken)
    {
        // await Task.Delay(10);
        
        _logger.LogInformation("ConsumerC starting");
        await Task.Run(() => _consumer.Subscribe(new List<string> { "topic" }));
        
        while (!cancellationToken.IsCancellationRequested)
        {
             _await Task.Run(() => consumer.Consume(cancellationToken));
        }
    }
user_35677360

When starting up the BackgroundServices, the framework is evidently doing something like this:

var starting1 = service1.ExecuteAsync(...); //all called in sequence without awaits inbetween
var starting2 = service2.ExecuteAsync(...);
var starting3 = service3.ExecuteAsync(...); 
...
//will await the startings all at once later on

Of course, when it does this in one of your services, it immediately gets trapped in a synchronous loop in which it blockingly polls the Kafka consumer. The thread of execution is never yielded back to the framework to continue calling other services.

You can get around this by doing your synchronous looping on separate threads, leaving the framework to happily go about its business:

protected Task ExecuteAsync(...) 
{
    return Task.Run(() => { //runs the below on a separate thread from the threadpool
        _logger.LogInformation("ConsumerC starting");
        _consumer.Subscribe(new List<string> { "topic" });

        while (!cancellationToken.IsCancellationRequested)
        {
            _ = _consumer.Consume(cancellationToken);
        }
    });
}

When dealing with async APIs, there's a general expectation that you don't sit on your given thread, as doing so can cause problems for things above you that are expecting the thread back. When you await things, the point of execution 'stays' in your code but really the thread is given back to the caller while a continuation is queued to carry on doing your stuff at the right time (transparently, mostly).

Unfortunately as far as I know the Kafka libraries don't have APIs for playing along with this, and so they require full threads of their own.

user_42526880
Related