kafka 客户端正在向代理失败的分区发送请求

Question

我正在使用 kafka-node 模块向 kafka 发送消息。 在集群环境中，我有一个具有 3 个分区和复制因子为 3 的主题。

主题描述是 -

Topic:clusterTopic      PartitionCount:3        ReplicationFactor:3    Configs:min.insync.replicas=2,segment.bytes=1073741824
        Topic: clusterTopic     Partition: 0    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3
        Topic: clusterTopic     Partition: 1    Leader: 2       Replicas: 2,3,1 Isr: 1,2,3
        Topic: clusterTopic     Partition: 2    Leader: 3       Replicas: 3,1,2 Isr: 1,2,3

生产者配置 -

        "requireAcks": 1,
        "attributes": 2,
        "partitionerType": 2,
        "retries": 2

当我发送数据时，它遵循像循环方式一样的循环（2）分区类型

当我按照以下步骤操作时

• 获取连接到 kafka:9092,kafka:9093 的 HighLevelProducer 实例
• 发消息
• 手动停止 kafka-server:9092
• 尝试使用 HighLevelProducer 发送另一条消息，然后 send() 将 触发回调错误：TimeoutError: Request timed out after 30000毫秒

我期望的是，如果一个分区不可访问（因为代理关闭），生产者应该自动将数据发送到下一个可用分区，但由于异常而我丢失了消息

异常如下-

  TimeoutError: Request timed out after 3000ms
    at new TimeoutError (\package\node_modules\kafka-node\lib\errors\TimeoutError.js:6:9)
    at Timeout.timeoutId._createTimeout [as _onTimeout] (\package\node_modules\kafka-node\lib\kafkaClient.js:980:14)
    at ontimeout (timers.js:424:11)
    at tryOnTimeout (timers.js:288:5)
    at listOnTimeout (timers.js:251:5)
    at Timer.processTimers (timers.js:211:10)
(node:56416) [DEP0079] DeprecationWarning: Custom inspection function on Objects via .inspect() is deprecated
  kafka-node:KafkaClient kafka-node-client reconnecting to kafka1:9092 +3s
  kafka-node:KafkaClient createBroker kafka1 9092 +1ms
  kafka-node:KafkaClient kafka-node-client reconnecting to kafka1:9092 +3s
  kafka-node:KafkaClient createBroker kafka1 9092 +0ms

Answer 1

请发送引导服务器进行确认，但根据手头的信息，我相信您遇到的情况如下：

• 您已将 min.insync.replicas 设置为 2
• 您已将 acks 设置为 1

通过这些设置，生产者会将事件发送到领导者副本并假设消息是安全的。

如果在发送后立即失败，并且在追随者赶上之前，您将丢失消息，因为您只等待一个确认。

但是，从代理的角度来看，您指定主题可用的要求是 2 个同步副本。默认情况下，仅允许同步的副本被选为领导者。由于第一个失败会导致关注者不同步，您的主题可能会被强制下线。您可以在测试中验证这一点，假设有一些设置。

要纠正，请尝试以下操作：

1. 如果高可用性最重要，请将 min.insync.replicas 设置为 1 并将acks 设置为 1
2. 如果不能接受数据丢失，请将 min.insync.replicas 设置为 2 并将 acks 设置为 all

您也可以将 unclean.leader.election.enable 设置为 true 以实现高可用性，因为这将允许不同步的副本被选为领导者，但这样就有可能丢失数据。