Consume Multiple KafkaTopics using KafkaFlow with Wild Card

[jcarpenter-greenmt]

I have a .NET application and want to consume multiple topics using the same consumer. I am reading my topic name in as an environment variable and want to include a wildcard (*) in order to consume from all matching topics.

The topic name I am using is like dev.std.*.fct.request.0. Looking at the KafkaFlow documentation on GitLab, it looks like the following should work when executed at startup in my ServiceCollectionExtensions.cs.

consumerConfigurationBuilder.Topic(kafkaBatchConsumerOptions.TopicName)
                            .WithConsumerConfig(new ConsumerConfig { ClientId = configuration["HOSTNAME"] })
                            .WithGroupId(kafkaBatchConsumerOptions.ConsumerGroupId)
                            .WithBufferSize(kafkaBatchConsumerOptions.BufferSize)

when kafkaBatchConsumerOptions.Topic name is set to the value above.

When I run this I get the following error.

[16:30:30 ERR] Kafka Consumer Error null
      Confluent.Kafka.ConsumeException: Broker: Unknown topic or partition
      at Confluent.Kafka.Consumer`2.Consume(Int32 millisecondsTimeout)
      at Confluent.Kafka.Consumer`2.Consume(CancellationToken cancellationToken)
      at KafkaFlow.Consumers.Consumer.ConsumeAsync(CancellationToken cancellationToken)

Am I missing something or is it not possible to set up a consumer in this way?

[jcarpenter-greenmt]

After a little more research I figured out how this works. I needed to convert my topic name to a regex string and it seems now all the topics are being consumed.

Setup code now looks like:

 var regexPattern = "^" + Regex.Escape(kafkaBatchConsumerOptions.TopicName).Replace("\\*", ".*") + "$";
        return consumerConfigurationBuilder => consumerConfigurationBuilder.Topic(regexPattern)
                                                                           .WithConsumerConfig(new ConsumerConfig { ClientId = configuration["HOSTNAME"] })
                                                                           .WithGroupId(kafkaBatchConsumerOptions.ConsumerGroupId)
                                                                           .WithBufferSize(kafkaBatchConsumerOptions.BufferSize)
                                                                           .WithWorkersCount(kafkaBatchConsumerOptions.WorkerCount)
                                                                           .WithAutoOffsetReset(AutoOffsetReset.Earliest)