rdkafka_group don't support regex pattern topics

[wcollin]

Describe the bug

rdkafka_group don't support regex pattern topics

To Reproduce

i use fluent/fluentd-kubernetes-daemonset:v1.19-debian-kafka2-1

Expected behavior

topics supports regex pattern like "@type kafka_group"

Your Environment

- Fluentd version: 1.19.0
- fluent-plugin-kafka version: 0.19.5
- ruby-kafka version: 0.22.2 x86_64-linux-gnu
- rdkafka version: 1.5.0

Your Configuration

<source>
  @type rdkafka_group
  @log_level info
  @label @toes

  kafka_configs {
    "bootstrap.servers": "#{ENV['FLUENTD_KAFKA_ADDR']}",
    "group.id": "logcenter_group",
    "auto.commit.interval.ms": "10000",
    "topic.metadata.refresh.interval.ms": "60000"
  }

  topics /test-*/
  add_prefix "kafka"
  format json
  retry_emit_limit 10

  # kafka consumer options
  max_wait_time_ms 500
  max_batch_size 10000
</source>

Your Error Log

2025-09-11 08:56:56 +0000 [error]: #0 unexpected error during consuming events from kafka. Re-fetch events. error="Broker: Unknown topic or partition (unknown_topic_or_part)"
  2025-09-11 08:56:56 +0000 [error]: #0 /fluentd/vendor/bundle/ruby/3.4.0/gems/rdkafka-0.22.2-x86_64-linux-gnu/lib/rdkafka/consumer.rb:550:in 'Rdkafka::Consumer#poll'
  2025-09-11 08:56:56 +0000 [error]: #0 /fluentd/vendor/bundle/ruby/3.4.0/gems/fluent-plugin-kafka-0.19.5/lib/fluent/plugin/in_rdkafka_group.rb:209:in 'Fluent::Plugin::RdKafkaGroupInput#each_batch'
  2025-09-11 08:56:56 +0000 [error]: #0 /fluentd/vendor/bundle/ruby/3.4.0/gems/fluent-plugin-kafka-0.19.5/lib/fluent/plugin/in_rdkafka_group.rb:229:in 'Fluent::Plugin::RdKafkaGroupInput#run'
  2025-09-11 08:56:56 +0000 [error]: #0 /fluentd/vendor/bundle/ruby/3.4.0/gems/fluentd-1.19.0/lib/fluent/plugin_helper/thread.rb:78:in 'block in Fluent::PluginHelper::Thread#thread_create'
2025-09-11 08:56:56 +0000 [warn]: #0 Stopping Consumer
2025-09-11 08:56:56 +0000 [warn]: #0 Could not connect to broker. retry_time:1. Next retry will be in 30 seconds

Additional context

No response

[Watson1978]

Hmm, I think this is the plugin's specification.

kafka_group uses ruby-kafka gem to communicate to kafka broker.
The library supports regexp pattern when subscribes the topics.
https://github.com/zendesk/ruby-kafka/blob/62a0cf9c44b0c05a4f98c4ec073ae5f04f4194e0/lib/kafka/consumer.rb#L110

However, rdkafka-ruby used in rdkafka_group does not support regexp pattern.
https://github.com/karafka/rdkafka-ruby/blob/35596902a1290dd23fb7845d918f3c827b951c94/lib/rdkafka/consumer.rb#L63

[wcollin]

@Watson1978 rdkafka-ruby support regexp pattern, hope rdkafka_group can support
karafka/rdkafka-ruby#556

[Watson1978]

Thanks.

https://github.com/confluentinc/librdkafka/blob/18f28bc8ccbd51abfc0d714de8716811755459b8/src/rdkafka.h#L4148-L4151

• Wildcard (regex) topics are supported:
• any topic name in the \p topics list that is prefixed with \c "^" will
• be regex-matched to the full list of topics in the cluster and matching
• topics will be added to the subscription list.
• The full topic list is retrieved every \c topic.metadata.refresh.interval.ms
• to pick up new or delete topics that match the subscription.
• If there is any change to the matched topics the consumer will
• immediately rejoin the group with the updated set of subscribed topics.
• Regex and full topic names can be mixed in \p topics.

Seems librdkafka supports wildcard (regex) topic subscriptions,
setting the topics parameter with a regular expression using String such as ^test_.* (it must starts with ^),
it allows Fluentd to consume messages from multiple matching topics automatically.

For example:

<source>
  @type rdkafka_group

  topics ^test_.*

...
</source>

[wcollin]

@Watson1978 thanks a lot, i will try it