kafka flusher native实现标准

[shalousun]

flusher配置参数

参数来自rdkafka，实现loongcollector native时按照当前社区的实现格式转换参数名

基础配置

参数名	类型	默认值	描述
bootstrap.servers	string	""	Kafka 集群的初始连接地址（逗号分隔，如 host1:port1,host2:port2）。
topic	string	""	topic。
message.max.bytes	int	1000000 (1MB)	单条消息的最大字节数（压缩前）。
receive.message.max.bytes	int	100000000 (100MB)	服务端能接收的单条消息最大字节数（需 ≥ message.max.bytes）。
queue.buffering.max.messages	int	100000	生产者队列中允许缓存的未发送消息的最大数量。该队列由所有 Topic 和 Partition 共享。如果设置为 0，则表示无限制（可能导致内存溢出）。
queue.buffering.max.kbytes	int	1048576 (1GB)	生产者队列中允许缓存的未发送消息的最大总字节数。该队列由所有 Topic 和 Partition 共享。此限制优先级高于 queue.buffering.max.messages（消息数量限制）
linger.ms	int	5	生产者发送消息前，允许消息在队列中累积的毫秒数。增大此值可提升批次处理效率（减少网络开销、优化压缩率），但会增加消息的传输延迟。
message.send.max.retries	int	2147483647	消息发送失败时的最大重试次数。注意：重试可能导致消息乱序，除非启用幂等性（enable.idempotence=true）。
retry.backoff.ms	int	1000	协议请求（如生产者消息发送、消费者拉取请求）重试时的最大退避等待时间。此值为指数退避（exponential backoff）策略的上限，用于避免高频重试导致系统过载。
batch.num.messages	int	10000	单个消息批次（MessageSet）中允许包含的最大消息数量。实际批次大小同时受 batch.size（字节限制）和 message.max.bytes（单条消息最大字节数）约束。
batch.size	int	1000000	单个消息批次（MessageSet）的最大字节数限制（包含协议帧开销）。该限制在首条消息加入批次后生效（无论首条消息大小），以确保超限消息仍可被发送。实际批次大小同时受batch.num.messages（消息数量限制）和message.max.bytes（单条消息最大字节数）约束。
acks	int	-1	控制生产者请求的副本确认机制： • 0：Leader无需返回响应，消息可能丢失。 • 1（默认）：Leader写入本地日志后立即响应，无需ISR副本确认。 • -1/all：Leader需等待所有ISR副本提交消息后才响应。若ISR副本数不足min.insync.replicas（Broker配置），请求将失败。
request.timeout.ms	int	30000 (30s)	生产者等待Broker响应ACK的超时时间（毫秒）。仅当request.required.acks（即acks）≠ 0时生效。超时后生产者会重试或抛出异常。
message.timeout.ms	int	300000 (5m)	生产者本地消息交付的最大等待时间（毫秒），包括重试。若超时或重试次数耗尽，触发交付失败。0表示无限等待。若配置了transactional.id，该值自动对齐transaction.timeout.ms。
partitioner	string	"consistent_random"	分区策略：random - 随机分配 consistent - 基于Key的CRC32哈希值分配（空Key和NULL Key会被固定分配到同一个分区） consistent_random - 基于Key的CRC32哈希值分配（空Key和NULL Key会被随机分配到不同分区） murmur2 - 与Java生产者兼容的Murmur2哈希分配（NULL Key会被固定分配到同一个分区） murmur2_random - 与Java生产者兼容的Murmur2哈希分配（NULL Key会被随机分配到不同分区，功能上与Java生产者的默认分区器等效） fnv1a - 基于Key的FNV-1a哈希值分配（NULL Key会被固定分配到同一个分区） fnv1a_random - 基于Key的FNV-1a哈希值分配（NULL Key会被随机分配到不同分区）
compression.type	string	"none"	压缩算法："none"、"gzip"、"snappy"、"lz4"、"zstd"。
compression.level	int	-1	压缩级别参数（对应配置属性compression.codec选定的算法）。数值越高压缩率越好，但会消耗更多CPU资源。可用范围取决于算法：gzip为[0-9]；lz4为[0-12]；snappy仅支持0；-1表示采用编解码器默认压缩级别。

权限认证

SASL认证

参数名	类型	默认值	描述
security.protocol	String	SASL_SSL	可选项plaintext, ssl, sasl_plaintext, sasl_ssl
sasl.mechanism	string	"PLAIN"	用于身份验证的SASL机制。支持的机制包括：GSSAPI、PLAIN、SCRAM-SHA-256、SCRAM-SHA-512、OAUTHBEARER。注意：尽管参数名为复数形式，但实际只需配置一种机制即可。
sasl.username	string	""	用于PLAIN和SASL-SCRAM机制的SASL用户名
sasl.password	String	""	用于PLAIN和SASL-SCRAM机制的SASL密码。

ssl认证

参数名	类型	默认值	描述
security.protocol	String	ssl	可选项plaintext, ssl, sasl_plaintext, sasl_ssl
enable.ssl.certificate.verification	Bool	True	启用 OpenSSL 内置的代理（服务器）证书验证。应用程序可以通过实现 certificate_verify_cb 来扩展此验证功能
ssl.certificate.location	string	""	客户端用于身份认证的公钥（PEM 格式）文件路径。
ssl.key.location	String	""	客户端用于身份认证的私钥（PEM 格式）文件路径
ssl.ca.location	string		用于验证服务端（Broker）密钥的CA证书文件或目录路径

Kerberos认证

参数名	类型	默认值	描述
security.protocol	String	SASL_SSL
sasl.mechanisms	string	"GSSAPI"	用于身份验证的SASL机制。支持的机制包括：GSSAPI、PLAIN、SCRAM-SHA-256、SCRAM-SHA-512、OAUTHBEARER。注意：尽管参数名为复数形式，但实际只需配置一种机制即可。Kerberos配置GSSAPI
sasl.kerberos.principal	String		客户端的 Kerberos 主体名称。（Windows 不支持此配置，将默认使用登录用户的 Kerberos 主体）
sasl.kerberos.service.name	String		服务名称，例如：kafka
sasl.kerberos.keytab	String		keytab文件路径
sasl.kerberos.kinit.cmd	String		用于刷新或获取客户端 Kerberos 票据的 Shell 命令，例如：kinit -R -t admin.keytab -k admin

kafka日志发送标准

在开源生态中95%的消息中间件消费组件都是基于单条消息来消费的处理的，并且各语言的sdk也提供了buffer来实现消息的批量发送，因此社区不实现多条日志合并打包成一条message的功能，在go版本flusher中也不提供这样的能力。