logstash kafka重复消费问题

L、先森

2019-09-30 0 4707

> Logstash消费一批数据时session timeout，导致offset没有提交给Kafka。解决方案是配置Logstash kafka input 的2个参数：max_poll_records 和 session_timeout_ms，增加 session timeout 或 降低 max poll records 或 二者都调整，让Logstash在允许的时间范围内把一批数据处理完。

[TOC]

#### Kafka输入配置选项

`auto_commit_interval_ms`

* 值类型为string
* 默认值为“5000”
* 消费者偏移量提交给Kafka的频率（毫秒）

`auto_offset_reset`

* 值类型为string
* 此设置没有默认值
* 如果Kafka中没有初始偏移量，或者偏移量超出范围，该怎么办：
    * `earliest`：自动重置偏移量到最早的偏移量
    * `latest`：自动重置偏移量到最新的偏移量
    * `none`：如果没有为消费者组找到先前的偏移量，则向消费者抛出异常
    * `其他`：向消费者抛出异常

`bootstrap_servers`

* 值类型为string
* 默认值为"localhost:9092"
* 用于建立到集群的初始连接的Kafka实例的url列表，这个列表应该是host1:port1,host2:port2的形式，这些url仅用于初始连接，以发现完整的集群成员（可能会动态更改），因此这个列表不需要包含完整的服务器集（不过，如果一个服务器宕机，你可能需要多个服务器）。

`check_crcs`

* 值类型为string
* 此设置没有默认值
* 自动检查被消费的记录的CRC32，这确保了在线路或磁盘上的消息没有发生损坏，这个检查增加了一些开销，因此在寻求极端性能的情况下可能会禁用它。

`client_id`

* 值类型为string
* 默认值为"logstash"
* 请求时要传递给服务器的id字符串，这样做的目的是通过允许包含逻辑应用程序名称来跟踪ip/端口以外的请求源。

`connections_max_idle_ms`

* 值类型为string
* 此设置没有默认值
* 在这个配置指定的毫秒数之后关闭空闲连接。

`consumer_threads`

* 值类型为number
* 默认值为1
* 理想情况下，为了达到完美的平衡，你应该拥有与分区数量一样多的线程，线程多于分区意味着有些线程将处于空闲状态。

`decorate_events`

* 值类型为boolean
* 默认值为false
* 可向事件添加Kafka元数据，比如主题、消息大小的选项，这将向logstash事件中添加一个名为kafka的字段，其中包含以下属性：topic：此消息关联的主题、consumer_group：这个事件中用来读取的消费者组、partition：此消息关联的分区、offset：此消息关联的分区的偏移量、key：包含消息key的ByteBuffer。

`enable_auto_commit`

* 值类型为string
* 默认值为"true"
* 如果是true，消费者定期向Kafka提交已经返回的消息的偏移量，当进程失败时，将使用这个提交的偏移量作为消费开始的位置。

`exclude_internal_topics`

* 值类型为string
* 此设置没有默认值
* 内部主题（如偏移量）的记录是否应该公开给消费者，如果设置为true，从内部主题接收记录的唯一方法就是订阅它。

`fetch_max_bytes`

* 值类型为string
* 此设置没有默认值
* 提取请求时服务器应该返回的最大数据量，这不是绝对最大值，如果提取的第一个非空分区中的第一个消息大于此值，消息仍然会返回，以确保消费者能够进行下去。

`fetch_max_wait_ms`

* 值类型为string
* 此设置没有默认值
* 如果没有足够的数据立即满足fetch_min_bytes，服务器在响应提取请求之前将阻塞的最大时间，这应该小于或等于poll_timeout_ms中使用的超时。

`fetch_min_bytes`

* 值类型为string
* 此设置没有默认值
* 提取请求时服务器应该返回的最小数据量，如果可用数据不足，请求将在响应请求前等待大量的数据积累。

`group_id`

* 值类型为string
* 默认值为"logstash"
* 此消费者所属的组的标识符，消费者组是由多个处理器组成的单个逻辑订阅服务器，主题中的消息将分发给具有相同group_id的所有Logstash实例。

`heartbeat_interval_ms`

* 值类型为string
* 此设置没有默认值
* 从心跳到消费者协调器的预期时间，心跳被用来确保消费者会话保持活跃，并在新消费者加入或离开组时促进重新平衡，该值必须设置为低于session.timeout.ms，但通常应该设置不高于该值的1/3，它可以调整得更低，以控制正常重新平衡的预期时间。

`jaas_path`

* 值类型为path
* 此设置没有默认值
* 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_config`

* 值类型为path
* 此设置没有默认值
* kerberos配置文件路径的选项，这是krb5.conf样式，详见https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html。

`key_deserializer_class`

* 值类型为string
* 默认值为"org.apache.kafka.common.serialization.StringDeserializer"
* 用于反序列化记录key的Java类。

`max_partition_fetch_bytes`

* 值类型为string
* 此设置没有默认值
* 服务器将返回每个分区的最大数据量，请求使用的最大总内存为#partitions * max.partition.fetch.bytes，这个大小必须至少是服务器允许的最大消息大小的最大值，否则可能生产者发送的消息比消费者能够提取的大，如果发生这种情况，消费者可能会陷入在某个分区上提取大量消息的困境。

`max_poll_interval_ms`

* 值类型为string
* 此设置没有默认值
* 当使用消费者组管理时，poll()调用之间的最大延迟，这为消费者在提取更多记录之前可以空闲的时间设置了上限，如果在超时过期之前没有调用poll()，就认为消费者失败了并且组将重新平衡，以便将分区重新分配给另一个成员，配置request_timeout_ms的值必须总是大于max_poll_interval_ms。

`max_poll_records`

* 值类型为string
* 此设置没有默认值
* 在对poll()的单个调用中返回的记录的最大数量。

`metadata_max_age_ms`

* 值类型为string
* 此设置没有默认值
* 以毫秒为单位的时间周期后，即使我们没有看到任何分区领导更改，也会强制刷新元数据，以主动发现任何新的broker或分区。

`partition_assignment_strategy`

* 值类型为string
* 此设置没有默认值
* 客户端将使用分区分配策略的类名在消费者实例之间分配分区所有权。

`poll_timeout_ms`

* 值类型为number
* 默认值为100
* kafka消费者将等待从主题接收新消息的时间。

`receive_buffer_bytes`

* 值类型为string
* 此设置没有默认值
* 读取数据时使用的TCP接收缓冲区（SO_RCVBUF）的大小

`reconnect_backoff_ms`

* 值类型为string
* 此设置没有默认值
* 在尝试重新连接到给定主机之前等待的时间量，这避免了在一个紧密循环中重复连接主机，此回退适用于消费者向broker发送的所有请求。

`request_timeout_ms`

* 值类型为string
* 此设置没有默认值
* 配置控制客户端等待请求响应的最长时间，如果超时之前没有收到响应，如有必要客户端将重新发送请求，或者在重试耗尽时失败请求。

`retry_backoff_ms`

* 值类型为string
* 此设置没有默认值
* 在试图重试失败的提取请求到给定主题分区之前等待的时间量，这避免了在一个紧凑的循环中重复的提取和失败。

`sasl_kerberos_service_name`

* 值类型为string
* 此设置没有默认值
* Kafka broker运行的Kerberos主体名称，这可以在Kafka的JAAS配置或Kafka的配置中定义。

`sasl_mechanism`

* 值类型为string
* 默认值为"GSSAPI"
* 用于客户端连接的SASL机制，这可能是安全提供者可用的任何机制，GSSAPI是默认机制。

`security_protocol`

* 值可以是：PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL
* 默认值为"PLAINTEXT"
* 要使用的安全协议，可以是PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL。

`send_buffer_bytes`

* 值类型为string
* 此设置没有默认值
* 发送数据时要使用的TCP发送缓冲区（SO_SNDBUF）的大小

`session_timeout_ms`

* 值类型为string
* 此设置没有默认值
* 超时之后，如果poll_timeout_ms没有被调用，消费者将被标记为死亡，并为group_id标识的组触发重新平衡操作。

`ssl_key_password`

* 值类型为password
* 此设置没有默认值
* 密钥存储文件中私有密钥的密码。

`ssl_keystore_location`

* 值类型为path
* 此设置没有默认值
* 如果需要客户端身份验证，则此设置存储密钥存储路径。

`ssl_keystore_password`

* 值类型为password
* 此设置没有默认值
* 如果需要客户端身份验证，则此设置存储密钥库密码

`ssl_keystore_type`

* 值类型为string
* 此设置没有默认值
* 密钥存储库类型。

`ssl_truststore_location`

* 值类型为path
* 此设置没有默认值
* JKS信任存储库路径用于验证Kafka broker的证书。

`ssl_truststore_password`

* 值类型为password
* 此设置没有默认值
* 信任存储库的密码。

`ssl_truststore_type`

* 
* 值类型为string
* 此设置没有默认值
* 信任存储库类型。

`topics`

* 值类型为array
* 默认值为["logstash"]
* 要订阅的主题列表，默认为["logstash"]。

`topics_pattern`

* 值类型为string
* 此设置没有默认值
* 订阅的主题正则表达式模式，使用此配置时，主题配置将被忽略。

`value_deserializer_class`

值类型为string
默认值为"org.apache.kafka.common.serialization.StringDeserializer"
用于反序列化记录值的Java类。

#### logstash 配置
```
# ------------ Pipeline Settings --------------
pipeline.workers: 2 
pipeline.output.workers: 1
pipeline.batch.size: 25
pipeline.batch.delay: 100
```
使用sleep控制每个event的处理时间
```
sleep {
    # Sleep 1 second for every event.
    time => "1"
}
```
#### 测试机器

| 核数 |内存  |
| --- | --- |
| 2 | 4 |
| 2 | 4 |

#### 单机
|  pipeline.workers| consumer_threads | session_timeout_ms|  max_poll_records| 结果 |
| --- | --- | --- | --- | --- |
|  1| 2| 25| 30 | 不重复 |
|  1| 2| 25| 40 | 不重复 |
|  1| 2| 25| 60 | 重复 |
|  2| 2| 25| 60 | 不重复 |
|  2| 2| 25| 80 | 不重复 |
|  2| 2| 25| 100 | 重复 |

#### 集群

|  pipeline.workers| consumer_threads | session_timeout_ms|  max_poll_records| 结果 |
| --- | --- | --- | --- | --- |
|  1| 1| 60| 20 | 不重复 |
|  1| 1| 60| 25 | 不重复 |
|  1| 1| 60| 30 | 重复 |
|  1| 2| 60| 10 | 不重复 |
|  1| 2| 120| 40 | 不重复 |
|  2| 1| 120| 160 | 重复 |
|  2| 1| 120| 140 | 不重复 |
|  2| 2| 120| 80 | 不重复 |
|  2| 2| 120| 90 | 重复 |
|  2| 2| 120| 100 | 重复 |
结论：
增加`pipeline.workers` 处理速度加倍,单机基本符合单线程max_poll_records = 每秒消费的条数 * session_timeout_ms，但集群空闲超过40秒，才不会出现重复即
`max_poll_records` * `consumer_threads`  要远小于 `pipeline.workers` * 每秒消费的条数 * `session_timeout_ms` 才不会出现重复

大数据

Java 8最快的垃圾搜集器是什么？

OpenJDK 8 有多种 GC（Garbage Collector）算法，如 Parallel GC、CMS 和 G1。哪一个才...

L、先森

10个实用的PHP正则表达式

正则表达式是程序开发中一个重要的元素，它提供用来描述或匹配文本的字符串，如特定的字符、词或算式等

L、先森

Spring知识点提炼

spring框架

L、先森