Kafka 输入插件

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

在 Kafka 中没有初始偏移量或偏移量超出范围时该怎么办

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

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

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

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

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 的基本身份验证的密码。

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

如果需要 Schema Registry 身份验证，此设置存储密钥库密码。

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

验证 Schema Registry 证书的信任库路径。

Schema Registry 信任库密码。

Schema Registry 信任库文件的格式。它必须是 jks 或 PKCS12。

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

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

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

要使用的安全协议，可以是 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。当您有两个或多个相同类型的插件时，这尤其有用，例如，如果您有两个 kafka 输入。在这种情况下添加命名 ID 将有助于在使用监控 API 监控 Logstash 时。

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

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

这可以帮助以后处理。

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

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

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

如果您尝试在已经有一个类型的事件上设置类型（例如，当您从发货器发送事件到索引器时），那么新的输入不会覆盖现有的类型。在发货器处设置的类型将与该事件保持关联，即使发送到另一台 Logstash 服务器也是如此。