node.js - 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: 请求超时30000毫秒

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

异常(exception)情况如下 -

  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

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

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

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

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

1. 如果高可用性最重要，请将 min.insync.replicas 设置为 1，将 ack 设置为 1
2. 如果数据丢失 Not Acceptable ，请将 min.insync.replicas 设置为 2 并确认全部

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