Kafka 输出插件

生产者在将请求视为已完成之前，要求领导者收到的确认次数。

acks=0。生产者不会等待服务器的任何确认。

acks=1。领导者会将记录写入其本地日志，但会在不等待所有跟随者的完全确认的情况下进行响应。

acks=all。领导者会等待完整的同步副本集，然后才确认记录。

当多个记录被发送到同一个分区时，生产者会尝试将记录批处理在一起，以减少请求次数。这有助于提高客户端和服务器的性能。此配置控制默认的批处理大小（以字节为单位）。

这是用于引导的，生产者只会将其用于获取元数据（主题、分区和副本）。用于发送实际数据的套接字连接将根据元数据中返回的代理信息建立。格式为 host1:port1,host2:port2，列表可以是代理的子集，也可以是指向代理子集的 VIP。

生产者可以使用多少字节的内存来缓冲等待发送到服务器的记录。

控制如何执行 DNS 查找。如果设置为 use_all_dns_ips，Logstash 会尝试所有为主机名返回的 IP 地址，然后再失败连接。如果设置为 resolve_canonical_bootstrap_servers_only，每个条目都会被解析并扩展为规范名称列表。

在发出请求时传递给服务器的 ID 字符串。这样做的目的是能够跟踪请求的来源，不仅仅是 ip/port，而是允许在请求中包含逻辑应用程序名称。

生产者生成的所有数据的压缩类型。默认值为 none（表示不压缩）。有效值为 none、gzip、snappy、lz4 或 zstd。

在由此配置指定的毫秒数后，关闭空闲连接。

Java 身份验证和授权服务 (JAAS) API 为 Kafka 提供用户身份验证和授权服务。此设置提供 JAAS 文件的路径。Kafka 客户端的 JAAS 文件示例

KafkaClient {
  com.sun.security.auth.module.Krb5LoginModule required
  useTicketCache=true
  renewTicket=true
  serviceName="kafka";
  };

请注意，在配置文件中指定 jaas_path 和 kerberos_config 会将它们添加到全局 JVM 系统属性中。这意味着如果您有多个 Kafka 输入，它们将共享相同的 jaas_path 和 kerberos_config。如果这不是您想要的，您需要在不同的 JVM 实例上运行 Logstash 的单独实例。

可选的 Kerberos 配置文件路径。这是 krb5.conf 样式，如 https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html 中所述。

消息键的序列化器类。

生产者将任何在请求传输之间到达的记录分组到单个批处理请求中。通常，这仅在负载下发生，此时记录到达的速度快于它们可以发送出去的速度。但是，在某些情况下，客户端可能希望即使在中等负载下也减少请求数量。此设置通过添加少量人工延迟来实现这一点，即，生产者不会立即发送记录，而是会等待最长指定的延迟，以允许发送其他记录，以便将发送批处理在一起。

请求的最大大小。

键值对的映射，每个对应于标头名称及其值。示例

    message_headers => { "event_timestamp" => "%{@timestamp}" }

消息的键。

获取主题元数据的初始元数据请求的超时设置。

强制刷新元数据之前的最大时间（以毫秒为单位）。

默认行为是散列事件的 message_key 以获取分区。当没有消息键时，插件以循环方式选择一个分区。

选择分区策略的可用选项如下

读取数据时使用的 TCP 接收缓冲区的大小。

连接失败后尝试重新连接到给定主机之前的等待时间。

该配置控制客户端等待请求响应的最长时间。如果在超时时间到期之前未收到响应，客户端将根据需要重新发送请求，或者如果重试次数用尽则失败请求。

默认重试行为是重试直到成功。为了防止数据丢失，不建议更改此设置。

如果您选择设置 retries，大于零的值将导致客户端仅重试固定次数。如果传输故障持续时间超过您的重试次数（网络中断、Kafka 停机等），这将导致数据丢失。

小于零的值是配置错误。

从版本 10.5.0 开始，此插件将仅重试是 RetriableException 和 InterruptException 子类的异常。如果生产消息抛出任何其他异常，则会记录错误，并且消息将被丢弃，不会重试。这可以防止 Logstash 管道无限期地挂起。

在 10.5.0 之前的版本中，任何异常都会无限期地重试，除非配置了 retries 选项。

尝试重新发送到给定主题分区的失败生产请求之前的等待时间。

此插件实例的本地 JAAS 配置设置，与使用 jaas_path 配置的配置文件设置相反，后者在整个 JVM 中共享。这允许每个插件实例都有自己的配置。

如果同时设置了 sasl_jaas_config 和 jaas_path 配置，则此处的设置优先。

示例（Azure 事件中心的设置）

    output {
      kafka {
        sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='auser'  password='apassword';"
      }
    }

Kafka 代理运行的 Kerberos 主体名称。这可以在 Kafka 的 JAAS 配置或 Kafka 的配置中定义。

用于客户端连接的 SASL 机制。这可以是任何安全提供程序可用的机制。GSSAPI 是默认机制。

要使用的安全协议，可以是 PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL 中的任何一个。

发送数据时使用的 TCP 发送缓冲区的大小。

端点标识算法，默认为 "https"。设置为空字符串 "" 以禁用

密钥库文件中的私钥的密码。

如果需要客户端身份验证，此设置将存储密钥库路径。

如果需要客户端身份验证，此设置将存储密钥库密码。

密钥库类型。

用于验证 Kafka 代理证书的 JKS 信任库路径。

信任库密码。

信任库类型。

要生产消息的主题

消息值的序列化器类

用于输出数据的编解码器。输出编解码器是方便地在数据离开输出之前对其进行编码的方法，而无需在 Logstash 管道中使用单独的过滤器。

禁用或启用此特定插件实例的指标日志记录。默认情况下，我们会记录所有可以记录的指标，但您可以禁用特定插件的指标收集。

在插件配置中添加一个唯一的 ID。如果未指定 ID，Logstash 将生成一个。强烈建议在配置中设置此 ID。当您有两个或多个相同类型的插件时，这尤其有用。例如，如果您有两个 kafka 输出。在这种情况下，添加一个命名 ID 将有助于在使用监控 API 时监控 Logstash。

output {
  kafka {
    id => "my_plugin_id"
  }
}