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，则定期向 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) 大小。

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

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

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

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 服务器也是如此。