Kafka Connect 将相同的任务分配给多个工作人员

Question

我在分布式模式下使用 Kafka Connect。我现在多次观察到的一个奇怪行为是，在一段时间后（可能是几小时，也可能是几天），似乎发生了平衡错误：相同的任务被分配给多个工人。因此，它们会同时运行，并且根据连接器的性质，会出现故障或产生“不可预测”的输出。

我能够用来重现该行为的最简单的配置是：两个 Kafka Connect 工作人员，两个连接器，每个连接器只有一个任务。 Kafka Connect 部署到 Kubernetes。 Kafka 本身位于 Confluent Cloud 中。 Kafka Connect 和 Kafka 的版本相同（5.3.1）。

日志中的相关消息：

工人甲：

[2019-10-30 12:44:23,925] INFO [Worker clientId=connect-1, groupId=some-kafka-connect-cluster] Successfully joined group with generation 488 (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:469)
[2019-10-30 12:44:23,926] INFO [Worker clientId=connect-1, groupId=some-kafka-connect-cluster] Joined group at generation 488 and got assignment: Assignment{error=0, leader='connect-1-d5c19893-b33c-4f07-85fb-db9736795759', leaderUrl='http://10.16.0.15:8083/', offset=250, connectorIds=[some-hdfs-sink, some-mqtt-source], taskIds=[some-hdfs-sink-0, some-mqtt-source-0], revokedConnectorIds=[], revokedTaskIds=[], delay=0} (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1397)

工人乙：

[2019-10-30 12:44:23,930] INFO [Worker clientId=connect-1, groupId=some-kafka-connect-cluster] Successfully joined group with generation 488 (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:469)
[2019-10-30 12:44:23,936] INFO [Worker clientId=connect-1, groupId=some-kafka-connect-cluster] Joined group at generation 488 and got assignment: Assignment{error=0, leader='connect-1-d5c19893-b33c-4f07-85fb-db9736795759', leaderUrl='http://10.16.0.15:8083/', offset=250, connectorIds=[some-mqtt-source], taskIds=[some-mqtt-source-0], revokedConnectorIds=[], revokedTaskIds=[], delay=0} (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1397)

在上面的日志摘录中，您可以观察到相同的任务 (some-mqtt-source-0) 分配给了两个工人。在这条消息之后，我还可以看到两个工作人员的任务实例的日志消息。

这种行为不依赖于连接器（我也在其他任务中观察到它）。它也不会在工人启动后立即发生，而只是在一段时间后发生。

我的问题是导致这种行为的原因是什么？

编辑 1： 我尝试运行 3 个工作人员，而不是两个，认为这可能是一个分布式共识问题。似乎不是，拥有 3 名工人并不能解决问题。

编辑 2： 我注意到，就在为工人 A 分配最初在工人 B 上运行的任务之前，该工人 (B) 观察到错误加入一个小组。例如，如果任务在第 N 代中“重复”，则工作人员 B 将不会在日志中显示“已成功加入第 N 代的组”消息。更重要的是，在 N-1 代和 N+1 代之间，worker B 通常会记录像 Attempt to heartbeat failed for since member id 和 Group coordinator bx-xxx-xxxxx.europe-west1.gcp.confluent.cloud:9092 (id: 1234567890 rack: null) is unavailable or invalid 这样的错误。 Worker B 通常在第 N 代之后不久加入第 N+1 代（有时仅在大约 3 秒后）。现在很清楚是什么触发了这种行为。然而：

• 虽然我了解可能存在此类临时问题，并且在一般情况下它们可能是正常的，为什么在所有服务器成功加入后重新平衡不能解决问题下一代？尽管随之而来的是更多的重新平衡 - 它没有正确分配任务，并且永远保持“重复”（直到我重新启动工作人员）。

• 似乎在某些时期，重新平衡几乎每几个小时发生一次，而在其他时期，它每 5 分钟发生一次（精确到几秒钟）；可能是什么原因？什么是正常的？

• 考虑到我使用 Confluent Cloud，“组协调器不可用或无效”错误的原因可能是什么，并且 是否有任何配置参数可以在 Kafka Connect 中调整以使它对这个错误更有弹性？我知道有session.timeout.ms 和heartbeat.interval.ms，但documentation 是如此简约，甚至不清楚将这些参数更改为更小或更大的值有什么实际影响。

编辑 3： 我观察到这个问题对于接收器任务并不重要：虽然相同的接收器任务被分配给多个工作人员，但相应的消费者被分配到通常应该的不同分区，并且一切都几乎按原样运行——我只是得到了比最初更多的任务要求。但是，在源任务的情况下，行为会中断 - 任务同时运行并在源端竞争资源。

编辑 4： 同时，我将 Kafka Connect 降级到 2.2 版（Confluent Platform 5.2.3）——"Incremental Cooperative Rebalancing" 之前的版本。它在过去 2 天工作正常。所以，我认为这种行为与新的再平衡机制有关。

Answer 1

正如 cmets 中提到的，Jira Kafka-9184 是为了解决这个问题，并且已经解决。

该修复适用于 2.3.2 及更高版本。

因此，现在的答案是：升级到最新版本应该可以防止此问题发生。