问题标签 [apache-kafka-mirrormaker]

MirrorMaker2 (strimzi:0.22.1-kafka-2.7.0) 配置为在启用组偏移同步的情况下从代理 A 到 B 进行一次定向镜像：

当源代理 A 上的一个节点在一段时间内不可用时，设置工作正常。从那时起，偏移量停止同步，并且延迟仅在目标代理 B 上增长，因为主题仅从源代理 A 消费。所以现在的问题是组偏移量停止同步。

到目前为止尝试了什么：

• 重新启动任务（无效）
• 重新启动 MirrorMaker2（无效）
• 重置目标代理 B 上的偏移量（偏移量没有同步，滞后只是在增长）
• 从目标代理 B 中删除组 ID（未在目标上重新创建组 ID）
• 删除主题（然后重新创建主题，而不是消费者组或偏移量开始同步）
• 禁用偏移同步并再次启用它（无效）

似乎破碎的消费者群体仍然存在于内部主题中，并阻止重新创建/重新同步这些主题。

有没有办法在不重新创建代理的情况下重新同步特定主题或 groupId？

作为 DR 计划的一部分，我正在实施 Kafka MirrorMaker 2.0 以在 2 个集群之间复制数据。

在这一点上，我只是想做一个简单的主动/被动设计。

这是我的 mm2.properties 文件：

问题在这里（我认为？）：

无论我如何配置流程，我只能看到正在写入west集群的数据。即使两个流都为真，west集群也会得到west.topic和central.topic。同时，central集群永远不会得到任何复制。

我还在两个集群中运行了这个 Mirror-Maker 配置，它总是west由于某种原因被写入集群。

我不明白为什么会发生这种情况，提前感谢您的帮助。

EDIT: I'm losing my mind a bit. I switched both URLs to be central like this:

and it STILL writes to the west cluster... how is this possible, when he doesn't even know where the west cluster is? I even checked the DNSs and they are correct.

最初，我创建了一个名为“quickstart-events”的主题，然后在其中生成了一些消息，然后使用消费者组“quickstartGroup”从 kafka-console-consumer 使用它，现在我想将该组从源复制到目标。

当我运行描述命令来描述源集群中的组时

我得到的输出是

在这里，主题正在被复制，但是当我运行命令来描述目标集群中的组时

得到错误为

My Mirror Maker 2 属性文件内容为：

PS：我使用的是 Kafka 2.8

我为 mirrormaker2 (mm2) 创建了 tls 配置并收到以下错误

MirrorSourceConnector 的错误调度程序在计划任务中捕获异常：刷新主题（org.apache.kafka.connect.mirror.Scheduler）[MirrorSourceConnector 刷新主题的调度程序] java.util.concurrent.ExecutionException：org.apache.kafka.common.errors .TimeoutException：在 org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java: 45）在 org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89) 在 org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32) 在 org. org.apache.kafka.connect.mirror.MirrorSourceConnector 上的 apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)。listTopics(MirrorSourceConnector.java:392) at org.apache.kafka.connect.mirror.MirrorSourceConnector.findSourceTopicPartitions(MirrorSourceConnector.java:194) at org.apache.kafka.connect.mirror.MirrorSourceConnector.refreshTopicPartitions(MirrorSourceConnector.java:218)在 org.apache.kafka.connect.mirror.Scheduler.run(Scheduler.java:93) 在 org.apache.kafka.connect.mirror.Scheduler.executeThread(Scheduler.java:112) 在 org.apache.kafka.connect .mirror.Scheduler.lambda$scheduleRepeatingDelayed$1(Scheduler.java:57) 在 java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) 在 java.base/java.util.concurrent。 FutureTask.runAndReset(FutureTask.java:305) 在 java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) 在 java.base/java.util.concurrent.ThreadPoolExecutor。runWorker(ThreadPoolExecutor.java:1128) 在 java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) 在 java.base/java.lang.Thread.run(Thread.java:829)

任何人都可以帮助澄清缺少什么吗？

在 Active-Active MM2 (MirrorMaker 2) 设置中，我有以下数据中心 (DC1,DC2)：

直流1：

有主题 T1

Kafka Streams 应用程序从 T1 消费并更新本地状态存储（例如：“MyStateStore”）。applicationId 是“myapp”。

DC2：

有主题 T1

MM2 进程将来自 DC1 中源 T1 的记录复制到 DC2 数据中心中的 DC1.T1 中。MM2 还将 myapp-MyStateStore-changelog 从 DC1 复制到 DC2 中的 DC1.myapp-MyStateStore-changelog。

MM2配置：

“sync.group.offsets.enabled = true”。

问题 ：

• 当我将“myapp”故障转移到 DC2 时，应用程序将如何访问 State Store？在我的代码中，我只能指定要从中使用的 statestore 的名称。
• 有没有办法让我也可以访问目标站点（DC2）中的 State Store？

我的理论是，在 DC2 中，只有名为 DC1.myapp-MyStateStore-changelog 的更改日志主题中有记录，因此当我将 myapp 故障转移到 DC2 时，所有状态都基本上丢失了。

我正在尝试设置 mirrormaker 。我已经将mirrormaker进程停靠了起来。我有两个 kafka 集群在同一台机器上运行。

卡夫卡码头工人在下面

第二个卡夫卡码头工人

如您所见，一个正在运行9092，另一个正在运行9093

我有consumer.properties如下

producer.properties如下_

当我尝试向 发布消息时kafka on 9092，出现以下错误

有人可以帮我解决这个问题。如果需要，请向我询问更多详细信息

我正在使用具有以下属性的 mm2 源（A），接收器（B）集群都有自己单独的动物园管理员我从源 A 中的主题测试中消耗一些数据。然后我停止了消费者，并在我将消费者指向同一组时启动镜像进程id 下沉然后它从头开始消费。我期待它应该从它在源中停止的地方开始。###############

我正在使用 SASL/SCRAM 运行 AWS MSK v2.7.0。我有一个运行 kafka 连接器和镜像制造商 2 (MM2) 的 ec2 实例。我能够在没有任何问题的情况下启动分发器，但是当我运行 curl 命令启动工作器时，输出看起来很好，但没有数据正在复制。svc_1 原则对源和目标都具有以下权限：

话题：

-read -create -describe -describe-configs

组：-describe -read

上述命令的输出：

另外，当我运行时：

我得到的只是[]。我也没有在日志中看到任何错误。有人可以让我知道我错过了什么或做错了什么吗？

谢谢，

集群 A 使用 JAAS 身份验证，集群 B 使用 Kerberos，尝试使用 connect mirrormaker 或 mirrormaker2 在集群 A 和 B 之间同步主题。进程成功启动并给出以下错误：

集群正在运行 K8s，mirrormaker 部署在集群 A 中的单独 kafka 客户端 pod 上。尝试使用 producer.properties 中的以下设置

• security.protocol=SASL_SSL
• sasl.mechanism=GSSAPI
• sasl.kerberos.service.name=kafka

有什么配置建议吗？

奇怪的是，在我启动mm2将数据从源集群A复制到目标集群B之后，在复制数据的同时，它也被删除了。见图：</p>

它最终删除了集群 B 中复制的所有数据。我不知道为什么会发生这种情况，有人可以帮助我吗？