Flink使用正则表达式消费多Kafka主题报错问题咨询

算力阿峰

2026-8-2

问题：Flink正则表达式消费Kafka主题失败排查

已知Flink支持通过正则表达式消费多个Kafka主题，现有Kafka主题示例如下：

sclee-10343434
sclee-10342432
sclee-34234
sclee-3343423432424
....

使用正则表达式sclee-[\d+]创建FlinkKafkaConsumer时抛出异常，代码如下：

val source = new FlinkKafkaConsumer[T](
  java.util.regex.Pattern.compile("sclee-[\\d+]"),
  deserializer,
  consumerProps
)

报错信息：

Caused by: java.lang.RuntimeException: Unable to retrieve any partitions with KafkaTopicsDescriptor: Topic Regex Pattern (dev-plexer-10507689[\d+])
    at org.apache.flink.streaming.connectors.kafka.internals.AbstractPartitionDiscoverer.discoverPartitions(AbstractPartitionDiscoverer.java:153)
    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.open(FlinkKafkaConsumerBase.java:553)
    at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
    at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:291)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:473)
    at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:92)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:469)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:522)
    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)
    at java.lang.Thread.run(Thread.java:750)

疑问：该正则表达式是否适用当前场景？Flink是否确实支持正则匹配多主题的功能？

解答

Flink确实支持通过正则表达式匹配多个Kafka主题的功能，你使用的1.4版本已明确该特性。
你当前使用的正则表达式sclee-[\d+]不适用场景：[\d+]是字符组，仅能匹配单个数字或加号，而你的主题是sclee-后跟随一串连续数字，因此该正则无法匹配任何现有主题，导致报错。
正确的正则表达式应为sclee-\d+：
- \d+表示匹配一个或多个连续数字，完全匹配你的主题格式。
- 在Scala代码中，由于Java的Pattern需要转义反斜杠，因此代码应修改为：
```
val source = new FlinkKafkaConsumer[T](
  java.util.regex.Pattern.compile("sclee-\\d+"),
  deserializer,
  consumerProps
)
```

内容的提问来源于stack exchange，提问作者sclee1