EMQX Confluent 数据集成桥接实战基于 Kafka 协议的 Confluent Producer 发布桥与 wolff/replayq 缓冲架构解析【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址: https://gitcode.com/gh_mirrors/em/emqx本文围绕 EMQX 开源仓库中 apps/emqx_bridge_confluent/README.md 所描述的核心应用展开讲解 EMQX Enterprise 中用于将消息写入 Confluent 云端的Confluent Producer 数据集成桥Data Integration Bridge它如何通过 Kafka 协议对接 Confluent、为何基于wolff库自带的replayq缓冲而无需emqx_resource缓冲工作进程、连接器Connector与动作Action的完整配置写法以及 OAuth 认证、SSL 强制启用等 Confluent 特有的细节。读完本文你将掌握在 EMQX 中配置 Confluent 数据出口的完整方法并理解其底层缓冲与连接管理的实现原理。应用定位EMQX 与 Confluent 之间的数据出口桥emqx_bridge_confluent是 EMQX 的一个桥接应用application其职责在 README 中定义得非常明确为EMQX Enterprise Edition提供Confluent Producer 数据集成桥通过Kafka 协议连接到 Confluent 云服务Confluent Cloud并发布消息属于数据出口Data Integration / Sink方向将 EMQX 中的 MQTT 消息桥接到 Confluent 的 Topic 中。从代码结构看该应用非常精简仅包含三个源文件与两个测试文件见 apps/emqx_bridge_confluent文件职责emqx_bridge_confluent_producer.erlHOCON 配置 Schema、示例配置、连接器配置转换emqx_connector_resource回调emqx_bridge_confluent_producer_action_info.erl动作信息注册emqx_action_info声明 action/connector 类型名与 Schema 模块emqx_bridge_confluent_producer_connector_info.erl连接器信息注册emqx_connector_info声明资源回调模块、Schema 模块与 API Schema两个 info 模块通过 mix.exs 中的emqx_action_info_modules/emqx_connector_info_modules环境变量注册到 EMQX使系统能够识别confluent_producer这一动作类型与连接器类型。值得注意的是Confluent 桥并不从零实现 Kafka 客户端而是复用 Kafka 桥的实现emqx_bridge_confluent_producer_connector_info中resource_callback_module()返回的是emqx_bridge_kafka_impl_producer配置 Schema 则通过 override emqx_bridge_kafka.erl 的字段得到。因此 Confluent 桥本质上可以理解为面向 Confluent Cloud 特化的 Kafka Producer 桥。架构要点wolff/replayq 自缓冲无需 emqx_resource 缓冲工作进程README 中最重要的技术说明是Currently, our Kafka Producer library (wolff) has its ownreplayqbuffering implementation, so this bridge does not require buffer workers fromemqx_resource.这段话包含两层关键事实缓冲由wolff库自带。EMQX 的 Kafka Producer 基于wolff库实现见 mix.exs 中的:wolff依赖而wolff内部使用replayq一种基于磁盘段的持久化队列完成消息缓冲与重放。因此该桥不需要emqx_resource框架提供的通用缓冲工作进程buffer worker也就绕开了emqx_resource_buffer_worker那套通用缓冲管线。不需要独立的 Connector 应用。README 还说明该桥直接实现了连接管理与交互因为 Confluent 桥不会被认证authentication与授权authorization应用使用所以无需像 Kafka 桥那样拆出单独的 connector app 来共享凭据/连接池。这一设计在底层实现中可以得到印证。在 emqx_bridge_kafka_impl_producer.erl 的producers_config/5约 L802-L857中EMQX 将 HOCON 配置翻译成wolff的 producer 配置其中包含replayq_dir、replayq_offload_mode、replayq_max_total_bytes、replayq_seg_bytes、drop_if_highmem等字段正是这些字段把 EMQX 的缓冲配置buffer 模式、每分区上限、段大小、内存过载保护传递给wolff/replayq。replayq的持久化目录位于 EMQX 数据目录下的kafka子目录中emqx_bridge_kafka_impl_producer.erl#L864-L872命名格式为bridge_type:bridge_name:node。代码中还实现了旧版本目录迁移逻辑maybe_migrate_old_replayq_dir/4L878-L896确保升级后磁盘缓冲数据可以平滑迁移避免积压消息丢失。从源码结构看还可以推断出两点设计考量因为缓冲能力在wolff内部连接器只有在真正启动 producer 后才会开始缓冲注释emqx_bridge_kafka_impl_producer.erl#L110-L114明确指出健康检查若处于connecting状态而非disconnected会给使用者连接器已就绪、可立即加动作并开始缓冲的错觉而 Kafka Producer 的缓冲只在 producer 启动后才生效这会导致数据丢失。因此健康检查流程特意区分了这两种状态。由于缓冲由wolff管理emqx_resource只负责资源的生命周期与健康检查消息投递走wolff的 batch API。Connector 配置连接 Confluent CloudConfluent 桥在 EMQX 中以连接器Connector 动作Action的两层模型管理。连接器负责与 Confluent Cloud 建立连接bootstrap 地址、认证、SSL、Socket 选项动作负责把消息写入具体 Topic。以下配置取自 emqx_bridge_confluent_tests.erl 中实际用于 Schema 校验的 HOCON 示例confluent_producer_connector_hocon/0L44-L63connectors.confluent_producer.my_producer { enable true authentication { username user password xxx } bootstrap_hosts xyz.sa-east1.gcp.confluent.cloud:9092 connect_timeout 5s metadata_request_timeout 5s min_metadata_refresh_interval 3s socket_opts { recbuf 1024KB sndbuf 1024KB tcp_keepalive none } }各字段说明结合 emqx_bridge_confluent_producer.erl 的 Schema 与 emqx_bridge_kafka.erl 的底层字段定义字段必填说明enable否是否启用该连接器默认trueauthentication是认证方式仅支持plainSASL/PLAIN 用户名密码或oauthOAuth Client Credentials由connector_overrides/0中的 union 类型约束L319-L333bootstrap_hosts是Confluent Cloud 的 bootstrap 地址列表如xyz.sa-east1.gcp.confluent.cloud:9092。解析时默认端口为9092且默认开启 SSRF 检查host_opts/0L436-L437connect_timeout否建立连接超时示例为5smetadata_request_timeout否元数据请求超时示例为5smin_metadata_refresh_interval否元数据最小刷新间隔示例为3ssocket_opts否TCP Socket 选项sndbuf/recbuf发送/接收缓冲示例1024KB、tcp_keepalive示例none等SSL 是强制的。与 Kafka 桥不同Kafka 默认ssl.enable falseConfluent 桥在 Schema 层面就把 SSL 默认改为开启%% Kafka has SSL disabled by default %% Confluent must use SSL ssl_overrides() - #{ enable mk(boolean(), #{default true}) }.见 emqx_bridge_confluent_producer.erl#L397-L402。同时连接器的ssl字段默认值为#{enable true}动作侧的默认值则进一步为#{enable true, verify verify_none}L367-L372。这意味着连接 Confluent Cloud 默认即走 TLS与 Confluent Cloud 仅暴露 TLS 端口的实际情况一致。示例连接器配置中还可显式指定ssl.versions [tlsv1.3, tlsv1.2]、server_name_indication、verify等参考 Schema 示例values({post, connector})L222-L235。认证机制SASL/PLAIN 与 OAuth Client CredentialsConfluent 桥的认证类型由connector_overrides限定为两种L319-L3331.plain用户名/密码auth_plain_overrides/0L382-L392将mechanism固定为plain并隐藏IMPORTANCE_HIDDEN强制username与password必填connectors.confluent_producer.my_confluent { authentication { mechanism plain # 由 Schema 固定无需显式写出 username ... password ... } }2.oauthOAuth Client Credentials这是 Confluent Cloud 推荐的方式Schema 中额外增加两个 Confluent 特有字段logical_cluster必填与identity_pool_id可选见 L89-L104。关键在于authentication_converter/2L439-L451当mechanism oauth时配置转换器会把logical_cluster重命名为logicalCluster、把identity_pool_id重命名为identityPoolId并合并进extensions映射最终随 OAuth token 请求一起发送给 Confluent。这一转换行为有专门的测试用例t_oauth_client_credentials_authn验证emqx_bridge_confluent_producer_SUITE.erl#L357-L396测试断言传给brod_oauth:auth/...的参数中extensions必须包含logicalCluster与identityPoolId且这些 Confluent 特有字段会覆盖任何已有扩展。示例配置含 OAuthconnectors.confluent_producer.my_confluent { bootstrap_hosts xyz.sa-east1.gcp.confluent.cloud:9092 authentication { mechanism oauth client_id ... client_secret ... token_endpoint ... logical_cluster confluent-logical-cluster # 必填 identity_pool_id confluent-identity-pool-id # 可选 } ssl { enable true } }Action 配置消息模板与 Kafka 生产参数动作Action定义了消息如何写入 Topic。以下 HOCON 同样取自 emqx_bridge_confluent_tests.erl 的 Schema 校验示例confluent_producer_action_hocon/0L14-L42actions.confluent_producer.my_producer { enable true connector my_connector parameters { buffer { memory_overload_protection false mode memory per_partition_limit 2GB segment_bytes 100MB } compression no_compression kafka_header_value_encode_mode none max_batch_bytes 896KB max_inflight 10 message { key ${.clientid} value ${.} } partition_count_refresh_interval 60s partition_strategy random query_mode async required_acks all_isr sync_query_timeout 5s topic test } }消息模板messagemessage.key与message.value使用 EMQX 的模板语法基于${...}占位符从 MQTT 消息中提取字段key ${.clientid}以客户端 ID 作为 Kafka 消息的 Key便于按客户端维度做分区路由value ${.}将整条 MQTT 消息含 payload、topic、clientid、qos、timestamp 等字段作为消息值写入 Kafka。与 Kafka 桥的一个差异是fields(kafka_message)从 Kafka 桥的字段中删除了timestampemqx_bridge_confluent_producer.erl#L117-L120说明 Confluent 桥不额外提供消息时间戳覆盖能力。消息头Kafka Headers可通过kafka_headers与kafka_ext_headers设置 Kafka 消息头kafka_header_value_encode_mode控制头值编码方式。Schema 示例values(action)L263-L297演示了如何将 MQTT 属性与自定义字段写入 Kafka 头parameters { kafka_headers ${.pub_props} # 将 MQTT 发布属性整体映射为 Kafka 头 kafka_ext_headers [ { kafka_ext_header_key clientid, kafka_ext_header_value ${clientid} } { kafka_ext_header_key topic, kafka_ext_header_value ${topic} } ] kafka_header_value_encode_mode none # 可选 none / base64 / ... }生产性能参数参数示例值说明max_linger_time5ms批内最大驻留时间linger控制吞吐与延迟的权衡max_linger_bytes10MB触发发送的累积字节上限max_batch_bytes896KB单个 Kafka 请求批次的最大字节数max_batch_ageinfinity默认批次最大年龄测试验证默认为infinity且可设置如500msmax_inflight10单个分区的在途请求数底层映射为wolff的max_send_ahead max_inflight - 1emqx_bridge_kafka_impl_producer.erl#L852max_retriesinfinity默认最大重试次数可设置为有限值如3reconnect_delay2s默认断线重连延迟测试验证默认为2000ms且可设置如1500mspartition_strategyrandom/key_dispatch分区策略key_dispatch按消息 Key 哈希路由底层为first_key_dispatchL859-L862partition_count_refresh_interval60s分区数刷新间隔配合动态主题使用required_acksall_isr生产确认级别all_isr表示等待所有 ISR 副本确认query_modeasync/sync查询模式同步模式配合sync_query_timeout示例5s使用compressionno_compression消息压缩方式可选snappy等依赖snappyer库见 mix.exspartitions_limitall_partitions默认最大分区数限制避免分区过多导致资源占用过高topictest目标 Topic支持t/${clientid}形式的动态主题模板缓冲参数buffer——wolff/replayq 的核心旋钮动作级parameters.buffer直接控制wolff内部replayq的行为。底层字段定义在 emqx_bridge_kafka.erl约 L542-L551默认值见 L167-L173字段默认值说明modememoryKafka 桥整体默认hybrid缓冲模式memory仅内存、disk仅磁盘、hybrid内存 磁盘卸载per_partition_limit256MBKafka 桥默认2GB每个分区允许缓冲的最大字节数底层映射为replayq_max_total_bytessegment_bytes10MB示例100MBreplayq 磁盘段文件大小底层映射为replayq_seg_bytesmemory_overload_protectiontrue内存过载保护底层映射为drop_if_highmem高内存时丢弃消息以保护节点mode与底层replayq配置的映射关系在 emqx_bridge_kafka_impl_producer.erl#L824-L836 中清晰可见memory→replayq_offload_mode false且不创建目录replayq_dir false纯内存缓冲disk→replayq_offload_mode false但使用磁盘目录全部落盘hybrid→replayq_offload_mode true并使用磁盘目录内存队列 磁盘段卸载。一个值得注意的约束producer_buffer_mode_validator/1emqx_bridge_kafka.erl#L838-L842会禁止磁盘模式 动态主题的组合disk模式配带占位符的 Topic 模板会校验失败因此动态主题只能配合memory或hybrid模式使用。测试套件中的t_disallow_disk_mode_for_dynamic_topic用例即验证了该限制emqx_bridge_confluent_producer_SUITE.erl#L354-L355。动态主题与多动作共享 TopicConfluent 桥完整继承了 Kafka 动作的能力动态主题topic支持模板如t/${clientid}由t_dynamic_topics测试用例验证L341-L352多动作共享同一 Topict_multiple_actions_sharing_topic验证了多个动作并发写同一 Topic 的场景L330-L339。这两个用例都通过emqx_bridge_kafka_action_SUITE的公共测试逻辑运行再次印证 Confluent 桥与 Kafka 桥共享同一套 producer 实现。另外测试套件中的t_same_name_confluent_kafka_bridgesL285-L328验证了 Confluent 桥与 Kafka 桥可以同名共存同一名称下分别创建confluent_producer与kafka_producer两类资源两者都能保持健康connected且禁用/启用 Kafka 桥不会影响同名 Confluent 桥继续投递消息。管理与 APIREST API连接器与动作均可通过 EMQX 管理 API 的GET/PUT/POST方法增删查改。Schema 中为每个 HTTP 方法定义了对应字段集合fields(get_connector)/put_connector/post_connector等见 L46-L74GET响应中会附加status如connected与node_status按节点报告连接状态等运行时信息values({get, ...})L185-L211。Schema 校验连接器与动作配置均通过hocon_tconf做严格校验。emqx_bridge_confluent_tests.erl中的测试确认scram_sha_256、scram_sha_512等 SASL 机制不合法matched_no_union_member只有plain与oauth可被接受L153-L176。健康检查t_on_get_status验证连接器/动作的状态上报失败时状态为connectingemqx_bridge_confluent_producer_SUITE.erl#L272-L274。集成测试环境CT 套件通过环境变量KAFKA_SASL_SSL_HOST/KAFKA_SASL_SSL_PORT指向带 SASL/SSL 的 Kafka 测试端点并使用toxiproxy做网络故障注入L28-L33模拟 Confluent Cloud 的 TLSSASL 环境。小结emqx_bridge_confluent是 EMQX 数据集成体系中面向 Confluent Cloud 的特化出口桥通过Kafka 协议发布消息到 Confluent复用emqx_bridge_kafka_impl_producer作为资源回调模块缓冲由wolff库的replayq承担内存/磁盘/混合三种模式支持持久化落盘与内存过载保护因此不依赖emqx_resource的通用缓冲工作进程由于不被认证/授权模块复用该桥直接内嵌连接管理无需独立 Connector 应用相比通用 Kafka 桥Confluent 桥在 Schema 层面强制 SSL 默认开启并支持 Confluent 特有的OAuth Client Credentials 认证logical_cluster/identity_pool_id自动映射为logicalCluster/identityPoolId扩展字段认证方式限定为plain与oauth两种。若需进一步深入可阅读 emqx_bridge_confluent_producer.erl 了解完整字段定义或参考 emqx_bridge_confluent_producer_SUITE.erl 与 emqx_bridge_confluent_tests.erl 中的真实配置示例与行为断言底层 producer 实现细节可追溯至 emqx_bridge_kafka_impl_producer.erl 与 emqx_bridge_kafka.erl。【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址: https://gitcode.com/gh_mirrors/em/emqx创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考