如何在Camel K的Kafka URL中使用${...}动态占位符？

问题根因 #

该问题由Apache Camel路由的静态端点解析机制导致，无需对美元符号或大括号做转义处理：

• Camel的.to()是静态端点方法，会在路由启动的初始化阶段一次性解析并固定Kafka的连接URI。此时还没有实际业务消息流入，无法读取消息头中的topic值，框架会直接把${header.topic}作为字面量的Topic名称传递给Kafka客户端，因此抛出非法Topic异常。
• .log()方法默认支持Simple表达式的运行时动态解析，每条消息流经时都会实时读取消息头的对应值渲染输出，因此可以正常打印出预期的Topic值。

解决方案 #

方案1（推荐，适用于Camel 2.16及以上版本） #

直接将静态.to()替换为专门用于动态解析端点URI的.toD()方法即可，修改后代码如下：

from(...)
.process(...)
.toD("kafka:${header.topic}?brokers=localhost:9092")

方案2（兼容Camel旧版本） #

使用收件人列表（Recipient List）集成模式实现动态路由：

from(...)
.process(...)
.recipientList(simple("kafka:${header.topic}?brokers=localhost:9092"))

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