集成层场景下，无需Kafka Stream用Kafka Connect+消息头实现动态路由可行吗？

完全可以在不依赖Kafka Streams的前提下，仅通过Kafka Connect实现基于消息头的动态Topic路由，所有逻辑都在Connect集群内部完成，无需额外的客户端应用。以下是具体实现方案：

方案1：使用内置SMT结合EL表达式（快速实现） #

Kafka Connect的Single Message Transform (SMT) 是原生支持的消息处理扩展，其中SetTopic SMT可以结合EL表达式从消息头中读取目标Topic，无需编写代码。

核心配置示例 #

假设你用FileStreamSourceConnector（实际可替换为任何Source Connector）消费源Topic，添加以下SMT配置即可实现动态路由：

# 源Connector基础配置
name=dynamic-router-source
connector.class=org.apache.kafka.connect.file.FileStreamSourceConnector
tasks.max=1
file=path/to/your/input/file
topic=source-topic

# SMT配置：从消息头获取目标Topic
transforms=routeToTarget
transforms.routeToTarget.type=org.apache.kafka.connect.transforms.SetTopic$Value
# 从名为target_topic的消息头取值，若不存在则默认发送到fallback-topic
transforms.routeToTarget.topic=${header:target_topic:fallback-topic}

这个配置会让Connect消费source-topic的消息后，根据消息头target_topic的值将消息转发到对应的目标Topic；如果该头字段缺失，消息会被发送到fallback-topic作为兜底。

方案2：自定义SMT（复杂路由逻辑） #

如果你的路由逻辑需要更复杂的判断（比如结合多个消息头字段、做业务规则校验等），可以自定义SMT来实现。

简单自定义SMT代码示例 #

import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.transforms.Transformation;
import org.apache.kafka.connect.header.Header;

import java.util.Map;

public class DynamicTopicRouter<R extends ConnectRecord<R>> implements Transformation<R> {

    @Override
    public void configure(Map<String, ?> configs) {
        // 可添加自定义配置参数
    }

    @Override
    public R apply(R record) {
        // 从消息头读取目标Topic
        Header targetHeader = record.headers().lastHeader("target_topic");
        if (targetHeader == null) {
            // 头字段缺失时的处理，比如返回原记录或抛异常（根据需求调整）
            return record;
        }
        String targetTopic = new String(targetHeader.value());
        
        // 这里可添加自定义路由逻辑，比如根据业务规则修改targetTopic
        if (targetTopic.startsWith("test-")) {
            targetTopic = "dev-" + targetTopic.substring(5);
        }

        // 返回修改了目标Topic的新记录
        return record.newRecord(
                targetTopic,
                record.kafkaPartition(),
                record.keySchema(),
                record.key(),
                record.valueSchema(),
                record.value(),
                record.timestamp()
        );
    }

    @Override
    public void close() {
        // 资源清理
    }
}

使用自定义SMT的步骤 #

1. 将上述代码打包成jar文件
2. 把jar放到Kafka Connect的插件目录（默认是share/java/kafka-connect下的对应文件夹，或通过plugin.path配置指定）
3. 在Connector配置中指定自定义SMT：

transforms=customRoute
transforms.customRoute.type=com.your.package.DynamicTopicRouter

关键注意事项 #

• 权限控制：确保Kafka Connect的worker账号拥有消费源Topic、生产所有可能的目标Topic的权限
• 错误处理：建议配置死信队列（DLQ），将路由失败的消息转发到指定Topic，避免丢失数据：

  errors.deadletterqueue.topic.name=router-dlq
  errors.deadletterqueue.context.headers.enable=true
  

• 性能考量：SMT在每个Connector任务的线程内同步执行，复杂逻辑可能影响吞吐量，建议做性能测试后调整任务数
• 版本兼容性：自定义SMT的依赖版本需与Kafka Connect的版本保持一致，避免出现依赖冲突

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