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表示采用编解码器默认压缩级别。

权限认证

rbkafka的权限认证配置项

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

权限部分实现目标

• 构建通用的Auth认证，无论是kafka、还是es都支持这些通用的tls、sasl、kerberos认证。
• 通用的Auth实现配置加载和验证。
• 其他各组建在集成通用认证配置时转映射成对应的sdk中的认证配置项。例如将Auth的配置构建成一个kafka sdk的映射map，然后完成sdk的配置设置，减少通过过多条件判断去设置注入到sdk的配置中。

多组件接入权限认证配置统一化

参数	类型	是否必选	默认值	说明
Authentication.TLS.Enabled	bool	否	false	启用 SSL 连接，对应 security.protocol=ssl
Authentication.TLS.CAFile	string	否	/	CA 证书路径，映射 ssl.ca.location
Authentication.TLS.CertFile	string	否	/	客户端证书路径，映射 ssl.certificate.location（与 KeyFile 必须成对配置，否则将视为配置错误）
Authentication.TLS.KeyFile	string	否	/	客户端私钥路径，映射 ssl.key.location（与 CertFile 必须成对配置，否则将视为配置错误）
Authentication.TLS.KeyPassword	string	否	/	私钥口令，映射 ssl.key.password（可选）
Authentication.Kerberos.Enabled	bool	否	false	启用 Kerberos（SASL/GSSAPI）
Authentication.Kerberos.Mechanisms	string	否	"GSSAPI"	映射 sasl.mechanisms
Authentication.Kerberos.ServiceName	string	否	"kafka"	映射 sasl.kerberos.service.name
Authentication.Kerberos.Principal	string	否	/	映射 sasl.kerberos.principal
Authentication.Kerberos.Keytab	string	否	/	映射 sasl.kerberos.keytab
Authentication.Kerberos.KinitCmd	string	否	/	映射 sasl.kerberos.kinit.cmd
Authentication.SASL.Mechanism	string	否	/	映射 sasl.mechanisms
Authentication.SASL.Username	string	否	/	映射 sasl.username
Authentication.SASL.Password	string	否	/	映射 sasl.password

AuthConfig 参考实现

下面的代码仅做实现指导，并未验证

文件：auth_config.h

// auth_config.h
#pragma once

#include <string>
#include <json/json.h>

class AuthConfig {
public:
    // =========================
    // 通用认证字段（值语义）
    // =========================

    // TLS/SSL
    bool tls_enabled = false;
    std::string tls_ca_file;
    std::string tls_cert_file;
    std::string tls_key_file;
    std::string tls_key_password;

    // Kerberos (GSSAPI)
    bool kerberos_enabled = false;
    std::string kerberos_mechanisms = "GSSAPI";
    std::string kerberos_service_name = "kafka";
    std::string kerberos_principal;
    std::string kerberos_keytab;
    std::string kerberos_kinit_cmd;

    // SASL (PLAIN, SCRAM, etc.)
    std::string sasl_mechanism;     // e.g., "PLAIN", "SCRAM-SHA-256"
    std::string sasl_username;
    std::string sasl_password;

    // =========================
    // 接口
    // =========================
    bool Load(const Json::Value& config, std::string& errorMsg);
    bool Validate(std::string& errorMsg) const;
};

---

文件：auth_config.cpp

// auth_config.cpp

#include "auth_config.h"
#include <algorithm>
#include <cctype>

// 工具：转小写
static std::string ToLower(const std::string& s) {
    std::string lower = s;
    std::transform(lower.begin(), lower.end(), lower.begin(), ::tolower);
    return lower;
}

// 通用 GetOptional 函数模板（简化实现）
template <typename T>
bool GetOptional(const Json::Value& obj, const char* key, T& value);

template <>
bool GetOptional<std::string>(const Json::Value& obj, const char* key, std::string& value) {
    if (obj.isMember(key) && obj[key].isString()) {
        value = obj[key].asString();
        return true;
    }
    return false;
}

template <>
bool GetOptional<bool>(const Json::Value& obj, const char* key, bool& value) {
    if (!obj.isMember(key)) return false;
    const auto& v = obj[key];
    if (v.isBool()) {
        value = v.asBool();
    } else if (v.isString()) {
        std::string sv = ToLower(v.asString());
        value = (sv == "true" || sv == "1" || sv == "yes");
    } else {
        return false;
    }
    return true;
}

// Load：只做字段提取，不关心映射
bool AuthConfig::Load(const Json::Value& config, std::string& errorMsg) {
    if (!config.isObject()) {
        errorMsg = "Authentication must be a JSON object";
        return false;
    }

    const Json::Value& auth = config;

    // TLS
    GetOptional(auth, "TLS.Enabled", tls_enabled);
    if (tls_enabled) {
        GetOptional(auth, "TLS.CAFile", tls_ca_file);
        GetOptional(auth, "TLS.CertFile", tls_cert_file);
        GetOptional(auth, "TLS.KeyFile", tls_key_file);
        GetOptional(auth, "TLS.KeyPassword", tls_key_password);
    }

    // Kerberos
    GetOptional(auth, "Kerberos.Enabled", kerberos_enabled);
    if (kerberos_enabled) {
        GetOptional(auth, "Kerberos.Mechanisms", kerberos_mechanisms);
        GetOptional(auth, "Kerberos.ServiceName", kerberos_service_name);
        GetOptional(auth, "Kerberos.Principal", kerberos_principal);
        GetOptional(auth, "Kerberos.Keytab", kerberos_keytab);
        GetOptional(auth, "Kerberos.KInitCmd", kerberos_kinit_cmd);
    }

    // SASL
    GetOptional(auth, "SASL.Mechanism", sasl_mechanism);
    GetOptional(auth, "SASL.Username", sasl_username);
    GetOptional(auth, "SASL.Password", sasl_password);

    return true;
}

// Validate：通用校验逻辑
bool AuthConfig::Validate(std::string& errorMsg) const {
    // TLS: CertFile 和 KeyFile 必须成对
    const bool has_cert = !tls_cert_file.empty();
    const bool has_key = !tls_key_file.empty();
    if (has_cert != has_key) {
        errorMsg = "Authentication.TLS: CertFile and KeyFile must be set together";
        return false;
    }

    // Kerberos: 启用时必须有 Principal 和 Keytab
    if (kerberos_enabled) {
        if (kerberos_principal.empty()) {
            errorMsg = "Authentication.Kerberos: Principal is required when Enabled=true";
            return false;
        }
        if (kerberos_keytab.empty()) {
            errorMsg = "Authentication.Kerberos: Keytab is required when Enabled=true";
            return false;
        }
    }

    // SASL: 有机制就必须有用户名密码
    if (!sasl_mechanism.empty()) {
        if (sasl_username.empty() || sasl_password.empty()) {
            errorMsg = "Authentication.SASL: Username and Password are required when Mechanism is set";
            return false;
        }
    }

    return true;
}

---

Kafka 组件：自行映射

// kafka_producer.cpp
bool KafkaProducer::Init(const Json::Value& config) {
    AuthConfig authConfig;
    if (config.isMember("Authentication")) {
        std::string errorMsg;
        if (!authConfig.Load(config["Authentication"], errorMsg)) {
            LOG_ERROR("Failed to load auth config: %s", errorMsg.c_str());
            return false;
        }

        if (!authConfig.Validate(errorMsg)) {
            LOG_ERROR("Invalid auth config: %s", errorMsg.c_str());
            return false;
        }
        // authConfig 映射到rd_kafka_conf
    }

    return true;
}

压缩配置参考说明

压缩算法	级别范围	默认级别	实现方式	特点	Kafka Broker 要求	适用场景
GZIP	0-9	Z_DEFAULT_COMPRESSION (通常为6)	zlib 的 deflateInit2()	高压缩率，CPU消耗大	无特殊要求	需要高压缩率的场景
LZ4	0-12	0 (无系统默认值)	LZ4F_preferences_t.compressionLevel	高速压缩/解压	无特殊要求	高吞吐量场景
ZSTD	0-22	3	ZSTD_initCStream()	平衡压缩率与速度	≥ 2.1.0	平衡型场景
Snappy	仅0	不适用	无级别控制	极速但压缩率低	无特殊要求	低延迟场景
None	不适用	不适用	无压缩	零开销	无要求	调试或非压缩场景

kafka日志发送标准

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