Cannot subscribe to kafka topics with regex patterns

[andreas-eberle]

I have a service that needs to listen to dynamically created kafka topics. All topics have a name like prefix.<UUID>.events. I tried using the following config in Quarkus' application.properties

mp.messaging.incoming.events.connector=smallrye-kafka
mp.messaging.incoming.events.topic=prefix\\..*\\.events
mp.messaging.incoming.events.pattern=true
mp.messaging.incoming.events.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
mp.messaging.incoming.events.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

However, on application startup I get the following errors

2020-08-18 17:37:06,490 ERROR [org.apa.kaf.cli.Metadata] (vert.x-kafka-consumer-thread-0) [Consumer clientId=consumer-fc9989b0-5c1c-4429-882c-a54589be2f04-1, groupId=fc9989b0-5c1c-4429-882c-a54589be2f04] Metadata response reported invalid topics [prefix\..*\.events]
2020-08-18 17:37:06,491 ERROR [io.sma.rea.mes.kafka] (vert.x-kafka-consumer-thread-0) SRMSG18217: Unable to read a record from Kafka topic 'prefix\..*\.events': org.apache.kafka.common.errors.InvalidTopicException: Invalid topics: [prefix\..*\.events]
2020-08-18 17:37:06,492 ERROR [io.sma.rea.mes.provider] (vert.x-kafka-consumer-thread-0) SRMSG00201: Error caught during the stream processing: org.apache.kafka.common.errors.InvalidTopicException: Invalid topics: [prefix\..*\.events]

I tested the pattern by running Pattern.compile("prefix\\..*\\.events") and it compiles and matches my topic names.

Furthermore, I tried using different patterns like .* and .+, which show the same error.

See also discussion on Zulip: https://quarkusio.zulipchat.com/#narrow/stream/187030-users/topic/Smallrye.20Reactive.20Messaging.3A.20Listen.20to.20dynamic.20Kafka.20topics

[cescoffier]

Seems to be related to quarkus and config.
A test using your regexp internally works.

For the record, I used the following MP Config file:

#file generated for testing purpose
#Wed Aug 19 19:57:32 CEST 2020
mp.messaging.incoming.kafka.topic=prefix\\..*\\.events
mp.messaging.incoming.kafka.pattern=true
mp.messaging.incoming.kafka.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
mp.messaging.incoming.kafka.connector=smallrye-kafka

[cescoffier]

I modified the Quarkus quickstart to use the following config:

# Configure the Kafka sink (we write to it)
mp.messaging.outgoing.generated-price.connector=smallrye-kafka
mp.messaging.outgoing.generated-price.topic=prefix.abc.events
mp.messaging.outgoing.generated-price.value.serializer=org.apache.kafka.common.serialization.IntegerSerializer

# Configure the Kafka source (we read from it)
mp.messaging.incoming.prices.connector=smallrye-kafka
mp.messaging.incoming.prices.topic=prefix\..*\.events
mp.messaging.incoming.prices.pattern=true
mp.messaging.incoming.prices.value.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer

and it works.

Can you check you use Quarkus 1.7?

[cescoffier]

Closing this issue. Please re-open if you still have the problem with 1.7.