Vector Kafka Sink将可观测数据发布到 Apache Kafka 的完整配置与实现指南【免费下载链接】vectorA high-performance observability data pipeline.项目地址: https://gitcode.com/GitHub_Trending/vect/vectorVector 的kafkasink 负责把日志、指标等可观测事件批量写入 Apache Kafka 主题。阅读本文你可以完整掌握该 sink 的全部配置参数bootstrap_servers、topic、key_field、encoding、batch、compression、sasl/tls、librdkafka_options等的含义、默认值与相互约束理解topic模板渲染与模板限制confinement机制以及 sink 底层如何通过librdkafka生产者、健康检查fetch_metadata和统计回调工作从而写出可复制、可运行、可验证的 Vector 配置。该组件在 Vector 中被标注为stable开发状态、投递保证为at_least_once至少一次无状态支持服务端健康检查与端到端确认acknowledgements。从 CUE 元数据看它接受 logs 以及各类 metrics 输入但不接受 traceswebsite/cue/reference/components/sinks/kafka.cue。组件特性概览kafkasink 的核心特性来自其 CUE 元数据定义投递保证at_least_once至少一次意味着事件可能被重复投递不会丢失。健康检查启用healthcheck.enabled: trueVector 在启动时对目标 topic 执行元数据拉取以验证连通性。确认机制支持acknowledgements用于事件级确认。发送能力支持批处理batch、压缩compression、编码encodingcodec可选json/text等以及 TLS。状态无状态stateful: false。sinks: kafka_sink: type: kafka inputs: [my_source] bootstrap_servers: 10.14.22.123:9092,10.14.23.332:9092 topic: topic-1234 encoding: codec: json上面的最小可运行配置只需type、bootstrap_servers、topic与encoding四个字段。其余均为可选用于调优批处理、压缩、认证与限流。完整配置参数说明下表汇总了kafkasink 的全部配置项描述、默认值与示例均来自仓库 CUE 定义website/cue/reference/components/sinks/generated/kafka.cue与源码KafkaSinkConfigsrc/sinks/kafka/config.rs。参数类型必填默认值说明bootstrap_serversstring是—逗号分隔的 Kafka 引导服务器格式host:port示例10.14.22.123:9092,10.14.23.332:9092topicstring模板是—要写入的主题名支持模板语法示例topic-1234、logs-{{unit}}-%Y-%m-%dhealthcheck_topicstring否使用topic健康检查专用主题当topic被模板化时可用固定值避免健康检查告警key_fieldstring路径否不发送 key用作分区 key 的日志字段名或 tag key示例user_id、.my_topic、%my_topicheaders_keystring路径否不写 headers用作 Kafka headers 的日志字段名旧别名headers_fieldencodingobject是—事件编码方式决定支持的输入类型logs / metrics / tracesbatchobject否无默认值批处理行为见下节compressionstring否none压缩算法none、gzip、snappy、lz4、zstdsaslobject否—SASL 认证配置enabled、username、password、mechanismtlsobject否—TLS 配置enabled、ca_file、crt_file、key_file、verify_certificate等socket_timeout_msuint毫秒否60000网络请求默认超时message_timeout_msuint毫秒否300000本地消息超时rate_limit_numuintrequests否i64::MAX极大值限流时间窗口内允许的最大请求数rate_limit_duration_secsuintseconds否1限流时间窗口长度librdkafka_optionsobject否空直接透传给底层librdkafka的键值对高级选项acknowledgementsobject否默认确认机制配置dangerously_allow_unconfined_template_resolutionbool否false关闭该 sink 所有模板限制检查危险见下节batch批处理行为batch子对象控制事件如何聚合成批。三个字段均可选timeout_secsfloat单位秒默认1.0批的最大存活时长超过即刷新。max_eventsuint单位 events批刷新前的最大事件数。max_bytesuint单位 bytes批的最大大小基于事件未压缩、未序列化的原始大小计算。关键点在于这些批处理选项最终会被映射为librdkafka的生产者选项而不是 Vector 层独立实现的批处理。从源码to_rdkafka()可以确认映射关系src/sinks/kafka/config.rsVector 的batch字段对应的librdkafka选项含义timeout_secsqueue.buffering.max.ms消息在发送前在队列中累积的延迟毫秒源值 ×1000max_eventsbatch.num.messages单个 MessageSet 中批量消息的最大数量max_bytesbatch.size单个 MessageSet 中所有消息批量化的最大字节数由于两者作用同一参数同时通过batch.*和librdkafka_options设置对应项会触发配置冲突并拒绝启动。该冲突在validate_batch_librdkafka_conflicts()中显式校验src/sinks/kafka/config.rs错误信息形如Batching setting batch.timeout_secs sets librdkafka_options.queue.buffering.max.ms.... The config already sets this ... Please delete one.。相关单元测试validate_rejects_batch_timeout_secs_conflicting_with_librdkafka_option等验证了这一点。batch: timeout_secs: 1.0 # 映射为 queue.buffering.max.ms 1000 max_events: 1000 # 映射为 batch.num.messages 1000 max_bytes: 1000000 # 映射为 batch.size 1000000librdkafka_options高级透传选项librdkafka_options是一个字符串键值对映射用于直接配置底层librdkafka客户端例如librdkafka_options: client.id: ${ENV_VAR} fetch.error.backoff.ms: 1000 socket.send.buffer.bytes: 100其值在to_rdkafka()的末尾被逐一写入ClientConfigsrc/sinks/kafka/config.rs。需要注意两点配置阶段validate只做纯映射、不构建原生生产者因此未知选项名与非法值在build阶段才被librdkafka原生配置拒绝。测试build_rejects_unknown_librdkafka_option与build_rejects_invalid_librdkafka_option印证了这一分阶段生命周期。若某个键同时被batch.*映射占用如queue.buffering.max.ms、batch.num.messages、batch.size会因上节所述冲突而被拒绝。compression压缩算法compression是一个字符串枚举源码定义在KafkaCompressionsrc/kafka.rs可选值与描述如下none默认不压缩。gzipGzip 压缩。snappySnappy 压缩。lz4LZ4 压缩。zstdZstandard 压缩。该值通过to_rdkafka()设置为librdkafka的compression.codecsrc/sinks/kafka/config.rs。topic 模板与限制confinementtopic字段支持模板语法例如logs-{{unit}}-%Y-%m-%d允许按事件字段或时间动态生成主题名。但出于安全考虑Vector 引入了**模板限制confinement**机制如果 topic 模板完全由不可信字段驱动如{{ topic }}默认会被拒绝以防止日志生产者改写任意目标主题。带静态前缀的模板如events-{{ env }}可以正常通过。完全无静态前缀的模板如{{ topic }}在默认配置下会被拒绝。设置dangerously_allow_unconfined_template_resolution: true可以全局关闭该 sink 的所有限制检查。CUE 中明确标注这是DANGEROUS — disables a security control开启后控制任意模板字段的日志生产者即可写入任意 key、路径或路由目标。在validate()中topic 模板会被confine()处理为ConfinedTemplatesrc/sinks/kafka/config.rs。对应的单元测试confinement_rejects_unconfined_topic、confinement_allows_prefixed_topic与confinement_opt_out_allows_unconfined_topic覆盖了三种情形。运行期每个事件都会先渲染 topic渲染失败会记录TemplateRenderingError并丢弃该事件src/sinks/kafka/sink.rs。认证SASL 与 TLSkafkasink 的认证配置由KafkaAuthConfig提供包含sasl与tls两个可选块src/kafka.rs。其核心逻辑在apply()方法中根据sasl.enabled与tls.enabled的组合决定security.protocolsrc/kafka.rssasl.enabledtls.enabledsecurity.protocolfalsefalseplaintextfalsetruessltruefalsesasl_plaintexttruetruesasl_ssl当启用 SASL 时username、password、mechanism分别映射为sasl.username、sasl.password、sasl.mechanism。文档示例给出的机制为SCRAM-SHA-256与SCRAM-SHA-512src/kafka.rs。源码注释特别说明通过sasl.*仅支持 PLAIN 与 SCRAM 系机制其他机制如 Kerberos必须通过librdkafka_options.*直接配置例如librdkafka_options.sasl.kerberos.service.name并且SASL 认证在 Windows 上不受支持。启用 TLS 时相关字段映射到librdkafka的 SSL 选项src/kafka.rsca_file/crt_file/key_file根据文件内容是否包含 PEM 起始标记分别设置为ssl.*.pem内联内容或ssl.*.location文件路径verify_hostname映射为ssl.endpoint.identification.algorithmhttps或noneverify_certificate映射为enable.ssl.certificate.verification。sasl: enabled: true mechanism: SCRAM-SHA-512 username: my_user password: my_secret tls: enabled: true verify_certificate: true ca_file: /path/to/ca.pem健康检查与底层生产者实现sink 构建时会创建一个FutureProducer并注入一个KafkaStatisticsContext上下文src/sinks/kafka/sink.rs。该上下文实现了ClientContext通过statistics.interval.ms 1000每秒接收一次librdkafka统计回调进而把统计量转换为 Vector 内部指标src/kafka.rs、src/sinks/kafka/config.rs。健康检查独立于主生产者运行healthcheck()函数创建一个新的BaseProducer然后对目标 topic 调用fetch_metadata()以验证可连通性src/sinks/kafka/sink.rs。当topic被模板化时健康检查若无法渲染出具体主题会记录告警此时可通过healthcheck_topic提供一个固定的检查主题来规避该告警。数据流本身由run_inner()组织src/sinks/kafka/sink.rs事件流先按key_field/headers_key/encoder 经KafkaRequestBuilder构建请求再通过带限流rate_limit_num/rate_limit_duration_secs由tower::limit::RateLimit实现的KafkaService驱动发送。Azure Event Hubs 复用从组件共享元数据website/cue/reference/components/kafka.cue可以看到kafkasink 也可用于连接 Azure Event HubsBasic tier 除外。典型配置为bootstrap_servers设为namespace.servicebus.windows.net:9093、sasl.enabled: true、sasl.mechanism: PLAIN、sasl.username: $$ConnectionString双$用于转义环境变量、sasl.password设为连接字符串并启用tls.enabled与tls.verify_certificate。完整配置示例下面是一个结合批处理、压缩、认证与限流的较完整配置示例sinks: kafka_sink: type: kafka inputs: [my_source] # 必填引导服务器与目标主题 bootstrap_servers: 10.14.22.123:9092,10.14.23.332:9092 topic: logs-{{unit}}-%Y-%m-%d healthcheck_topic: logs-healthcheck # topic 模板化时的固定检查主题 # 分区 key 与 headers均可选 key_field: user_id headers_key: headers # 编码 encoding: codec: json # 批处理映射为 librdkafka 生产者选项 batch: timeout_secs: 1.0 max_events: 1000 max_bytes: 1000000 # 压缩 compression: lz4 # 认证 sasl: enabled: true mechanism: SCRAM-SHA-512 username: my_user password: my_secret tls: enabled: true verify_certificate: true # 超时与限流 socket_timeout_ms: 60000 message_timeout_ms: 300000 rate_limit_num: 1000 rate_limit_duration_secs: 1 # 高级透传选项避免与 batch.* 冲突 librdkafka_options: client.id: vector-sink遥测指标该 sink 通过KafkaStatisticsReceived内部事件暴露一组 Kafka 相关指标src/internal_events/kafka.rs、website/cue/reference/components/sinks/kafka.cuekafka_queue_messages/kafka_queue_messages_bytes生产者队列中待发送消息数与字节数。kafka_requests_total/kafka_requests_bytes_total发出的请求数与字节数。kafka_responses_total/kafka_responses_bytes_total收到的响应数与字节数。kafka_produced_messages_total/kafka_produced_messages_bytes_total已产生生产成功的消息数与字节数。kafka_consumed_messages_total/kafka_consumed_messages_bytes_total已消费的消息数与字节数。这些指标每秒由librdkafka统计回调刷新可直接用于监控 sink 的吞吐、队列积压与错误情况。小结kafkasink 以librdkafka为底层生产者将 Vector 的batch、compression、sasl/tls等高层配置映射为librdkafka生产者选项从而在保持至少一次投递保证的同时提供批处理、压缩、限流与健康检查能力。配置时需要特别注意batch.*与librdkafka_options的键冲突校验、topic模板的 confinement 限制以及 SASL/TLS 组合对security.protocol的决定作用。以上参数、默认值与实现细节均可在 src/sinks/kafka/config.rs、src/kafka.rs 与 website/cue/reference/components/sinks/generated/kafka.cue 中查证。【免费下载链接】vectorA high-performance observability data pipeline.项目地址: https://gitcode.com/GitHub_Trending/vect/vector创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Vector Kafka Sink：将可观测数据发布到 Apache Kafka 的完整配置与实现指南
Vector Kafka Sink将可观测数据发布到 Apache Kafka 的完整配置与实现指南【免费下载链接】vectorA high-performance observability data pipeline.项目地址: https://gitcode.com/GitHub_Trending/vect/vectorVector 的kafkasink 负责把日志、指标等可观测事件批量写入 Apache Kafka 主题。阅读本文你可以完整掌握该 sink 的全部配置参数bootstrap_servers、topic、key_field、encoding、batch、compression、sasl/tls、librdkafka_options等的含义、默认值与相互约束理解topic模板渲染与模板限制confinement机制以及 sink 底层如何通过librdkafka生产者、健康检查fetch_metadata和统计回调工作从而写出可复制、可运行、可验证的 Vector 配置。该组件在 Vector 中被标注为stable开发状态、投递保证为at_least_once至少一次无状态支持服务端健康检查与端到端确认acknowledgements。从 CUE 元数据看它接受 logs 以及各类 metrics 输入但不接受 traceswebsite/cue/reference/components/sinks/kafka.cue。组件特性概览kafkasink 的核心特性来自其 CUE 元数据定义投递保证at_least_once至少一次意味着事件可能被重复投递不会丢失。健康检查启用healthcheck.enabled: trueVector 在启动时对目标 topic 执行元数据拉取以验证连通性。确认机制支持acknowledgements用于事件级确认。发送能力支持批处理batch、压缩compression、编码encodingcodec可选json/text等以及 TLS。状态无状态stateful: false。sinks: kafka_sink: type: kafka inputs: [my_source] bootstrap_servers: 10.14.22.123:9092,10.14.23.332:9092 topic: topic-1234 encoding: codec: json上面的最小可运行配置只需type、bootstrap_servers、topic与encoding四个字段。其余均为可选用于调优批处理、压缩、认证与限流。完整配置参数说明下表汇总了kafkasink 的全部配置项描述、默认值与示例均来自仓库 CUE 定义website/cue/reference/components/sinks/generated/kafka.cue与源码KafkaSinkConfigsrc/sinks/kafka/config.rs。参数类型必填默认值说明bootstrap_serversstring是—逗号分隔的 Kafka 引导服务器格式host:port示例10.14.22.123:9092,10.14.23.332:9092topicstring模板是—要写入的主题名支持模板语法示例topic-1234、logs-{{unit}}-%Y-%m-%dhealthcheck_topicstring否使用topic健康检查专用主题当topic被模板化时可用固定值避免健康检查告警key_fieldstring路径否不发送 key用作分区 key 的日志字段名或 tag key示例user_id、.my_topic、%my_topicheaders_keystring路径否不写 headers用作 Kafka headers 的日志字段名旧别名headers_fieldencodingobject是—事件编码方式决定支持的输入类型logs / metrics / tracesbatchobject否无默认值批处理行为见下节compressionstring否none压缩算法none、gzip、snappy、lz4、zstdsaslobject否—SASL 认证配置enabled、username、password、mechanismtlsobject否—TLS 配置enabled、ca_file、crt_file、key_file、verify_certificate等socket_timeout_msuint毫秒否60000网络请求默认超时message_timeout_msuint毫秒否300000本地消息超时rate_limit_numuintrequests否i64::MAX极大值限流时间窗口内允许的最大请求数rate_limit_duration_secsuintseconds否1限流时间窗口长度librdkafka_optionsobject否空直接透传给底层librdkafka的键值对高级选项acknowledgementsobject否默认确认机制配置dangerously_allow_unconfined_template_resolutionbool否false关闭该 sink 所有模板限制检查危险见下节batch批处理行为batch子对象控制事件如何聚合成批。三个字段均可选timeout_secsfloat单位秒默认1.0批的最大存活时长超过即刷新。max_eventsuint单位 events批刷新前的最大事件数。max_bytesuint单位 bytes批的最大大小基于事件未压缩、未序列化的原始大小计算。关键点在于这些批处理选项最终会被映射为librdkafka的生产者选项而不是 Vector 层独立实现的批处理。从源码to_rdkafka()可以确认映射关系src/sinks/kafka/config.rsVector 的batch字段对应的librdkafka选项含义timeout_secsqueue.buffering.max.ms消息在发送前在队列中累积的延迟毫秒源值 ×1000max_eventsbatch.num.messages单个 MessageSet 中批量消息的最大数量max_bytesbatch.size单个 MessageSet 中所有消息批量化的最大字节数由于两者作用同一参数同时通过batch.*和librdkafka_options设置对应项会触发配置冲突并拒绝启动。该冲突在validate_batch_librdkafka_conflicts()中显式校验src/sinks/kafka/config.rs错误信息形如Batching setting batch.timeout_secs sets librdkafka_options.queue.buffering.max.ms.... The config already sets this ... Please delete one.。相关单元测试validate_rejects_batch_timeout_secs_conflicting_with_librdkafka_option等验证了这一点。batch: timeout_secs: 1.0 # 映射为 queue.buffering.max.ms 1000 max_events: 1000 # 映射为 batch.num.messages 1000 max_bytes: 1000000 # 映射为 batch.size 1000000librdkafka_options高级透传选项librdkafka_options是一个字符串键值对映射用于直接配置底层librdkafka客户端例如librdkafka_options: client.id: ${ENV_VAR} fetch.error.backoff.ms: 1000 socket.send.buffer.bytes: 100其值在to_rdkafka()的末尾被逐一写入ClientConfigsrc/sinks/kafka/config.rs。需要注意两点配置阶段validate只做纯映射、不构建原生生产者因此未知选项名与非法值在build阶段才被librdkafka原生配置拒绝。测试build_rejects_unknown_librdkafka_option与build_rejects_invalid_librdkafka_option印证了这一分阶段生命周期。若某个键同时被batch.*映射占用如queue.buffering.max.ms、batch.num.messages、batch.size会因上节所述冲突而被拒绝。compression压缩算法compression是一个字符串枚举源码定义在KafkaCompressionsrc/kafka.rs可选值与描述如下none默认不压缩。gzipGzip 压缩。snappySnappy 压缩。lz4LZ4 压缩。zstdZstandard 压缩。该值通过to_rdkafka()设置为librdkafka的compression.codecsrc/sinks/kafka/config.rs。topic 模板与限制confinementtopic字段支持模板语法例如logs-{{unit}}-%Y-%m-%d允许按事件字段或时间动态生成主题名。但出于安全考虑Vector 引入了**模板限制confinement**机制如果 topic 模板完全由不可信字段驱动如{{ topic }}默认会被拒绝以防止日志生产者改写任意目标主题。带静态前缀的模板如events-{{ env }}可以正常通过。完全无静态前缀的模板如{{ topic }}在默认配置下会被拒绝。设置dangerously_allow_unconfined_template_resolution: true可以全局关闭该 sink 的所有限制检查。CUE 中明确标注这是DANGEROUS — disables a security control开启后控制任意模板字段的日志生产者即可写入任意 key、路径或路由目标。在validate()中topic 模板会被confine()处理为ConfinedTemplatesrc/sinks/kafka/config.rs。对应的单元测试confinement_rejects_unconfined_topic、confinement_allows_prefixed_topic与confinement_opt_out_allows_unconfined_topic覆盖了三种情形。运行期每个事件都会先渲染 topic渲染失败会记录TemplateRenderingError并丢弃该事件src/sinks/kafka/sink.rs。认证SASL 与 TLSkafkasink 的认证配置由KafkaAuthConfig提供包含sasl与tls两个可选块src/kafka.rs。其核心逻辑在apply()方法中根据sasl.enabled与tls.enabled的组合决定security.protocolsrc/kafka.rssasl.enabledtls.enabledsecurity.protocolfalsefalseplaintextfalsetruessltruefalsesasl_plaintexttruetruesasl_ssl当启用 SASL 时username、password、mechanism分别映射为sasl.username、sasl.password、sasl.mechanism。文档示例给出的机制为SCRAM-SHA-256与SCRAM-SHA-512src/kafka.rs。源码注释特别说明通过sasl.*仅支持 PLAIN 与 SCRAM 系机制其他机制如 Kerberos必须通过librdkafka_options.*直接配置例如librdkafka_options.sasl.kerberos.service.name并且SASL 认证在 Windows 上不受支持。启用 TLS 时相关字段映射到librdkafka的 SSL 选项src/kafka.rsca_file/crt_file/key_file根据文件内容是否包含 PEM 起始标记分别设置为ssl.*.pem内联内容或ssl.*.location文件路径verify_hostname映射为ssl.endpoint.identification.algorithmhttps或noneverify_certificate映射为enable.ssl.certificate.verification。sasl: enabled: true mechanism: SCRAM-SHA-512 username: my_user password: my_secret tls: enabled: true verify_certificate: true ca_file: /path/to/ca.pem健康检查与底层生产者实现sink 构建时会创建一个FutureProducer并注入一个KafkaStatisticsContext上下文src/sinks/kafka/sink.rs。该上下文实现了ClientContext通过statistics.interval.ms 1000每秒接收一次librdkafka统计回调进而把统计量转换为 Vector 内部指标src/kafka.rs、src/sinks/kafka/config.rs。健康检查独立于主生产者运行healthcheck()函数创建一个新的BaseProducer然后对目标 topic 调用fetch_metadata()以验证可连通性src/sinks/kafka/sink.rs。当topic被模板化时健康检查若无法渲染出具体主题会记录告警此时可通过healthcheck_topic提供一个固定的检查主题来规避该告警。数据流本身由run_inner()组织src/sinks/kafka/sink.rs事件流先按key_field/headers_key/encoder 经KafkaRequestBuilder构建请求再通过带限流rate_limit_num/rate_limit_duration_secs由tower::limit::RateLimit实现的KafkaService驱动发送。Azure Event Hubs 复用从组件共享元数据website/cue/reference/components/kafka.cue可以看到kafkasink 也可用于连接 Azure Event HubsBasic tier 除外。典型配置为bootstrap_servers设为namespace.servicebus.windows.net:9093、sasl.enabled: true、sasl.mechanism: PLAIN、sasl.username: $$ConnectionString双$用于转义环境变量、sasl.password设为连接字符串并启用tls.enabled与tls.verify_certificate。完整配置示例下面是一个结合批处理、压缩、认证与限流的较完整配置示例sinks: kafka_sink: type: kafka inputs: [my_source] # 必填引导服务器与目标主题 bootstrap_servers: 10.14.22.123:9092,10.14.23.332:9092 topic: logs-{{unit}}-%Y-%m-%d healthcheck_topic: logs-healthcheck # topic 模板化时的固定检查主题 # 分区 key 与 headers均可选 key_field: user_id headers_key: headers # 编码 encoding: codec: json # 批处理映射为 librdkafka 生产者选项 batch: timeout_secs: 1.0 max_events: 1000 max_bytes: 1000000 # 压缩 compression: lz4 # 认证 sasl: enabled: true mechanism: SCRAM-SHA-512 username: my_user password: my_secret tls: enabled: true verify_certificate: true # 超时与限流 socket_timeout_ms: 60000 message_timeout_ms: 300000 rate_limit_num: 1000 rate_limit_duration_secs: 1 # 高级透传选项避免与 batch.* 冲突 librdkafka_options: client.id: vector-sink遥测指标该 sink 通过KafkaStatisticsReceived内部事件暴露一组 Kafka 相关指标src/internal_events/kafka.rs、website/cue/reference/components/sinks/kafka.cuekafka_queue_messages/kafka_queue_messages_bytes生产者队列中待发送消息数与字节数。kafka_requests_total/kafka_requests_bytes_total发出的请求数与字节数。kafka_responses_total/kafka_responses_bytes_total收到的响应数与字节数。kafka_produced_messages_total/kafka_produced_messages_bytes_total已产生生产成功的消息数与字节数。kafka_consumed_messages_total/kafka_consumed_messages_bytes_total已消费的消息数与字节数。这些指标每秒由librdkafka统计回调刷新可直接用于监控 sink 的吞吐、队列积压与错误情况。小结kafkasink 以librdkafka为底层生产者将 Vector 的batch、compression、sasl/tls等高层配置映射为librdkafka生产者选项从而在保持至少一次投递保证的同时提供批处理、压缩、限流与健康检查能力。配置时需要特别注意batch.*与librdkafka_options的键冲突校验、topic模板的 confinement 限制以及 SASL/TLS 组合对security.protocol的决定作用。以上参数、默认值与实现细节均可在 src/sinks/kafka/config.rs、src/kafka.rs 与 website/cue/reference/components/sinks/generated/kafka.cue 中查证。【免费下载链接】vectorA high-performance observability data pipeline.项目地址: https://gitcode.com/GitHub_Trending/vect/vector创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
