Kafka 输入插件

将消费者偏移量提交到 Kafka 的频率（以毫秒为单位）。

当 Kafka 中没有初始偏移量或偏移量超出范围时该如何处理

用于建立与集群初始连接的 Kafka 实例的 URL 列表。此列表应采用 host1:port1,host2:port2 的形式。这些 URL 仅用于初始连接以发现完整的集群成员资格（可能会动态更改），因此此列表不必包含完整的服务器集（但您可能需要多个，以防服务器宕机）。

自动检查消费的记录的 CRC32。这可以确保消息未发生在线路或磁盘上的损坏。此检查会增加一些开销，因此在寻求极致性能的情况下可能会禁用它。

应如何执行 DNS 查找。如果设置为 use_all_dns_ips，则当查找返回主机名的多个 IP 地址时，会在连接失败之前尝试连接所有 IP 地址。如果该值为 resolve_canonical_bootstrap_servers_only，则会解析每个条目并将其扩展为规范名称列表。

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

Kafka 消费者的机架标识符。用于选择消费者从中读取的物理上最近的机架。该设置与 Kafka 的 broker.rack 配置相对应。

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

理想情况下，您应该拥有与分区数一样多的线程以实现完美的平衡，即线程数多于分区数意味着某些线程将处于空闲状态。

可选项，用于将 Kafka 元数据（如主题、消息大小和头部键值）添加到事件中。这会在 Logstash 事件中添加一个名为 kafka 的字段，其中包含以下属性

  * 值类型是 布尔值 * 默认值是 true

控制在订阅不存在的主题时是否自动创建该主题。只有当此配置设置为 true 并且代理上使用 auto.create.topics.enable 启用了自动创建主题时，才会自动创建主题；否则不允许自动创建主题。

当进程失败时，此提交的偏移量将用作开始消费的位置。

如果为 true，则定期将消费者已返回的消息的偏移量提交到 Kafka。如果值为 false，则每次消费者将从主题获取的数据写入内存或持久队列时，都会提交偏移量。

是否应将来自内部主题（例如偏移量）的记录公开给消费者。如果设置为 true，则接收来自内部主题的记录的唯一方法是订阅该主题。

服务器应该为获取请求返回的最大数据量。这不是绝对的最大值，如果获取请求的第一个非空分区中的第一条消息大于此值，则仍将返回该消息，以确保消费者可以取得进展。

如果数据不足以立即满足 fetch_min_bytes，则服务器在响应获取请求之前将阻塞的最大时间量。该值应小于或等于 poll_timeout_ms 中使用的超时时间。

服务器应该为获取请求返回的最小数据量。如果可用数据不足，则请求将等待积累这么多数据，然后再响应请求。

此消费者所属的组的标识符。消费者组是由多个处理器组成的单个逻辑订阅者。主题中的消息将分发给所有具有相同 group_id 的 Logstash 实例。

此 Logstash Kafka 消费者的静态成员身份标识符。静态成员资格功能在 KIP-345 中引入，在 Kafka 属性 group.instance.id 下可用。其目的是避免在消费者下线后必须转发大量数据的情况下进行重新平衡。此功能可以缓解服务状态繁重的情况，其中从实例 A 到 B 的一个主题分区的重新平衡会导致大量数据传输。频繁下线/上线的客户端可以使用此选项避免频繁和繁重的重新平衡。

向消费者协调器发送心跳的预期时间间隔。心跳用于确保消费者的会话保持活动状态，并在新消费者加入或离开组时促进重新平衡。该值必须设置得低于 session.timeout.ms，但通常应设置为该值的 1/3 或更低。可以将其调整得更低，以控制正常重新平衡的预期时间。

控制如何读取以事务方式写入的消息。如果设置为 read_committed，则轮询消息将仅返回已提交的事务性消息。如果设置为 read_uncommitted（默认值），则轮询消息将返回所有消息，即使是已中止的事务性消息。在任一模式下，都将无条件返回非事务性消息。

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 配置文件的可选路径。这是 https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html 中详细介绍的 krb5.conf 样式。

用于反序列化记录键的 Java 类

服务器将返回的每个分区的最大数据量。请求使用的最大总内存将为 #partitions * max.partition.fetch.bytes。此大小必须至少与服务器允许的最大消息大小一样大，否则生产者可能会发送大于消费者可以获取的消息。如果发生这种情况，消费者可能会在尝试获取某个分区上的大消息时卡住。

在使用消费者组管理时，调用 poll() 的最大延迟时间。这为消费者在获取更多记录之前可以空闲的时间设置了上限。如果未在此超时到期之前调用 poll()，则认为消费者失败，并且该组将重新平衡，以便将分区重新分配给另一个成员。

在单个 poll() 调用中返回的最大记录数。

即使我们没有看到任何分区领导权的更改，也强制刷新元数据的时间间隔（以毫秒为单位），以便主动发现任何新的代理或分区。

客户端用于在消费者实例之间分配分区所有权的分区分配策略的名称，支持的选项包括

这些映射到 Kafka 相应的 ConsumerPartitionAssignor 实现。

Kafka 消费者将等待从主题接收新消息的时间。

在订阅一组主题后，Kafka 消费者会在轮询时自动加入该组。插件循环中的轮询可确保消费者处于活动状态。在底层，Kafka 客户端会向服务器发送定期心跳信号。超时指定在每次轮询时等待输入阻塞的时间。

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

尝试重新连接到给定主机之前等待的时间。这避免了在紧密循环中重复连接到主机。此退避应用于消费者发送给代理的所有请求。

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

在尝试重试对给定主题分区的失败的提取请求之前等待的时间。这避免了在紧密循环中重复提取和失败。

指定的 SASL 机制应使用的 SASL 客户端回调处理程序类。

此插件实例本地的 JAAS 配置设置，而不是使用 jaas_path 配置的配置文件设置，这些设置在 JVM 之间共享。这允许每个插件实例拥有自己的配置。

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

示例（Azure 事件中心设置）

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

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

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

设置用于访问远程 Schema Registry 的基本身份验证的用户名。

设置转发 HTTP 代理的地址。空字符串被视为未设置代理。

设置用于访问远程 Schema Registry 的基本身份验证的密码。

如果需要架构注册表客户端身份验证，则此设置存储密钥库路径。

如果需要架构注册表身份验证，则此设置存储密钥库密码。

密钥库文件的格式。它必须是 jks 或 PKCS12。

用于验证架构注册表证书的信任存储路径。

架构注册表信任存储密码。

架构注册表信任存储文件的格式。它必须是 jks 或 PKCS12。

指向 Schema Registry 服务实例的 URI，用于管理 Avro 架构。请确保已将用于反序列化指定主题数据的 Avro 架构上传到 Schema Registry 服务。架构必须遵循模式为 <主题名称>-value 的命名约定。

使用 Schema Registry 配置选项或 value_deserializer_class 配置选项，但不能同时使用两者。

当使用架构注册表时，默认情况下，插件会在注册期间、处理事件之前检查连接并验证架构注册表。在某些情况下，当它尝试验证已通过身份验证的架构注册表时，此过程可能会失败，从而导致插件崩溃。此设置允许插件在注册期间跳过验证，从而允许插件继续并处理事件。请注意，配置错误的架构注册表仍会阻止插件处理事件。

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

发送数据时要使用的 TCP 发送缓冲区 (SO_SNDBUF) 的大小

如果未调用 poll_timeout_ms，则超过此超时时间后，消费者将被标记为死，并为 group_id 标识的组触发重新平衡操作

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

密钥存储文件中私钥的密码。

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

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

密钥库文件的格式。它必须是 jks 或 PKCS12。

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

信任存储密码。

信任存储文件的格式。它必须是 jks 或 PKCS12。

要订阅的主题列表，默认为 ["logstash"]。

要订阅的主题正则表达式模式。

通过从代理检索完整的主题名称列表并在本地应用模式来完成正则表达式过滤。当与具有大量主题的代理一起使用时，此操作可能非常缓慢，尤其是在有很多消费者的情况下。

用于反序列化记录值的 Java 类。仅当您不使用 Schema Registry 时，才能使用自定义值反序列化器。使用 value_deserializer_class 配置选项或 schema_registry_url 配置选项，但不能同时使用两者。

向事件添加字段

用于输入数据的编解码器。输入编解码器是在数据进入输入之前解码数据的便捷方法，而无需在 Logstash 管道中单独的过滤器。

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

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

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

向您的事件添加任意数量的任意标签。

这可以帮助稍后进行处理。

向此输入处理的所有事件添加 type 字段。

类型主要用于过滤器激活。

该类型作为事件本身的一部分存储，因此您也可以使用该类型在 Kibana 中搜索它。

如果您尝试在已经具有类型的事件上设置类型（例如，当您从发件人向索引器发送事件时），则新的输入不会覆盖现有类型。在发件人处设置的类型会与该事件的生命周期保持一致，即使将其发送到另一个 Logstash 服务器也是如此。