kafka集成需五步：一配cli-config.yaml声明连接参数；二用hermes skill run提交flink作业；三执行hermes status验证kafka/flink状态；四用kafka-console-producer注入测试消息；五通过telnet、权限检查等排查连接失败。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜
如果您正在使用Hermes Agent构建事件驱动型应用，但无法将传感器数据或业务事件可靠地接入Kafka，则可能是由于连接配置缺失、消费者组状态异常或Topic未正确声明所致。以下是完成Kafka集成的实操步骤：
一、配置Kafka连接参数
该步骤用于在Hermes Agent运行时环境中显式声明Kafka集群的访问入口与消费上下文，确保Agent能识别并建立到Broker的稳定长连接。
1、定位项目根目录下的cli-config.yaml.example文件。
2、将文件重命名为cli-config.yaml。
3、在文件中添加kafka配置区块，内容如下：
4、设置bootstrap_servers为Kafka集群实际监听地址，例如"localhost:9092"。
5、指定待消费的topic名称，例如"iot_sensor_data"。
6、为消费者分配唯一consumer_group标识，例如"hermes_consumer"。
二、启动Flink流处理任务
该步骤通过Hermes Agent命令行工具提交预定义的Flink作业，使Agent具备从Kafka拉取原始消息、执行实时计算并输出结果的能力。
1、确保Flink集群已启动且可被Hermes Agent访问。
2、确认./config/flink_job.yaml中已定义source（Kafka connector）、transform（如窗口聚合）和sink（如数据库或告警服务）。
3、执行以下命令提交作业：
4、命令中flink:iot_analytics为内置技能标识符，代表物联网分析流水线。
5、若返回Job submitted successfully，表示Flink任务已进入运行态。
三、验证Kafka端到端数据流转
该步骤利用Hermes Agent内置的状态检查工具，确认Kafka连接器是否完成初始化、消费者是否成功加入组并开始拉取消息。
1、运行状态检查命令：./hermes status --service kafka,flink。
2、观察输出中kafka服务项的status字段是否为"connected"。
3、检查consumer_group字段是否显示当前活跃成员数大于0。
4、确认latest_offset与current_offset差值持续收敛，表明无积压。
5、若任一指标异常，需检查Kafka Broker日志中是否存在UNKNOWN_MEMBER_ID或NOT_COORDINATOR错误。
四、手动注入测试消息验证链路
该步骤绕过生产者SDK，直接使用Kafka原生命令行工具向目标Topic写入轻量测试载荷，快速验证整个接收通路是否就绪。
1、进入Kafka安装目录下的bin/子目录。
2、执行生产者命令：./kafka-console-producer.sh --bootstrap-server localhost:9092 --topic iot_sensor_data。
3、在终端中输入JSON格式测试消息，例如：{"device_id":"D001","temp":23.5,"ts":1745084820}。
4、按下回车立即发送，无需额外提交操作。
5、切换至Hermes Agent日志输出窗口，搜索关键字"received message"，确认消息已被捕获并解析。
五、排查常见连接失败原因
该步骤提供针对Kafka连接中断的多路径诊断手段，覆盖网络、认证、权限三个关键层面。
1、使用telnet localhost 9092测试Broker端口连通性，若失败则检查防火墙或Docker网络配置。
2、若Kafka启用SASL_SSL认证（如华为云ROMA Connect MQS），需在cli-config.yaml中补充sasl_jaas_config与security_protocol字段。
3、登录Kafka管理控制台，确认iot_sensor_data Topic已创建且状态为"active"。
4、检查Topic权限策略，确保hermes_consumer所属用户具有Describe与Read操作权限。
5、在cli-config.yaml中临时将auto_offset_reset设为"earliest"，排除因offset越界导致的静默跳过。
