【问题标题】：How to migrate consumer offsets using MirrorMaker 2.0?如何使用 MirrorMaker 2.0 迁移消费者偏移量？
【发布时间】：2021-07-27 19:31:06
【问题描述】：

对于 Kafka 2.7.0，我使用 MirroMaker 2.0 作为 Kafka 连接连接器，将所有主题从主 Kafka 集群复制到备份集群。

除了__consumer_offsets，所有主题都被完美复制。以下是连接配置：

{
    "name": "test-connector",
    "config": {
      "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
      "topics.blacklist": "some-random-topic",
      "replication.policy.separator": "",
      "source.cluster.alias": "",
      "target.cluster.alias": "",
      "exclude.internal.topics":"false",
      "tasks.max": "10",
      "key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
      "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
      "source.cluster.bootstrap.servers": "xx.xx.xxx.xx:9094",
      "target.cluster.bootstrap.servers": "yy.yy.yyy.yy:9094",
      "topics": "test-topic-from-primary,primary-kafka-connect-offset,primary-kafka-connect-config,primary-kafka-connect-status,__consumer_offsets"
    }
}

在类似的问题here 中，接受的答案如下：

在你的 consumer.config 中添加这个：

exclude.internal.topics=false

并将其添加到您的 producer.config 中：

client.id=__admin_client

我应该在我的配置中的哪里添加这些？

这里Connector Configuration Properties没有client.id这样的属性，不过我已经将exclude.internal.topics的值设置为false。

这里有什么我遗漏的吗？

更新

我了解到 Kafka 2.7 及更高版本支持使用 MirrorCheckpointTask 自动同步消费者偏移量，如 here 所述。

我为此创建了一个具有以下配置的连接器：

{
    "name": "mirror-checkpoint-connector",
    "config": {
      "connector.class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
      "sync.group.offsets.enabled": "true",
      "source.cluster.alias": "",
      "target.cluster.alias": "",
      "exclude.internal.topics":"false",
      "tasks.max": "10",
      "key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
      "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
      "source.cluster.bootstrap.servers": "xx.xx.xxx.xx:9094",
      "target.cluster.bootstrap.servers": "yy.yy.yyy.yy:9094",
      "topics": "__consumer_offsets"
    }
}

仍然没有帮助。这是正确的方法吗？有什么需要吗？

【问题讨论】：

标签： apache-kafka kafka-consumer-api apache-kafka-connect confluent-platform apache-kafka-mirrormaker

【解决方案1】：

您不想复制 consumer_offsets。由于各种原因，从 src 到目标集群的偏移量不会相同。

MirrorMaker2 提供了进行偏移平移的能力。它将使用从 src 集群生成的转换后的偏移量填充目标集群。 https://cwiki.apache.org/confluence/display/KAFKA/KIP-545%3A+support+automated+consumer+offset+sync+across+clusters+in+MM+2.0

【讨论】：

所以为了实现这一点，我需要做的就是将"sync.group.offsets.enabled" : "true" 添加到配置中？
是的。在 Kafka 2.7 及更高版本中。
我使用的是 Kafka 2.7.0，但这对我不起作用。我有"topics": "test-topic,__consumer_offsets"，在这个test-topic 被复制但__consumer_offsets 被忽略。

【解决方案2】：

__consumer_offsets 默认被忽略

topics.exclude = [.*[\-\.]internal, .*\.replica, __.*]

您需要覆盖此配置

【讨论】：