apache-kafka - Kafka MirrorMaker 2.0复制每个消息

Question

我正在尝试使用MirrorMaker 2.0复制Kafka集群。我正在使用以下mm2.properties:

name = mirror-site1-site2
topics = .*
connector.class = org.apache.kafka.connect.mirror.MirrorSourceConnector
tasks.max = 1
plugin.path=/usr/share/java/kafka/plugin
clusters = site1, site2

# for demo, source and target clusters are the same
source.cluster.alias = site1
target.cluster.alias = site2

site1.sasl.mechanism=SCRAM-SHA-256
site1.security.protocol=SASL_PLAINTEXT
site1.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
   username="<someuser>" \
   password="<somepass>";

site2.sasl.mechanism=SCRAM-SHA-256
site2.security.protocol=SASL_PLAINTEXT
site2.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
   username="<someuser>" \
   password="<somepass>";

site1.bootstrap.servers = <IP1>:9093, <IP2>:9093, <IP3>:9093, <IP4>:9093
site2.bootstrap.servers = <IP5>:9093, <IP6>:9093, <IP7>:9093, <IP8>:9093

site1->site2.enabled = true
site1->site2.topics = topic1


# use ByteArrayConverter to ensure that records are not re-encoded
key.converter = org.apache.kafka.connect.converters.ByteArrayConverter
value.converter = org.apache.kafka.connect.converters.ByteArrayConverter

所以这是问题所在，mm2似乎总是复制x3消息:

# Manual message production: 

 kafkacat -P -b <IP1>:9093,<IP2>:9093,<IP3>:9093,<IP4>:9093 -t "topic1"


# Result in the source topic (site1 cluster): 

% Reached end of topic topic1 [2] at offset 405
Message1
% Reached end of topic topic1 [2] at offset 406
Message2
% Reached end of topic topic1 [6] at offset 408
Message3
% Reached end of topic topic1 [2] at offset 407

 kafkacat -P -b <IP5>:9093,<IP6>:9093,<IP7>:9093,<IP8>:9093 -t "site1.topic1"

# Result in the target topic (site2 cluster): 

% Reached end of topic site1.titi [2] at offset 1216
Message1
Message1
Message1
% Reached end of topic site1.titi [2] at offset 1219
Message2
Message2
Message2
% Reached end of topic site1.titi [6] at offset 1229
Message3
Message3
Message3

我尝试使用来自融合软件包的Kafka和直接来自Apache的kafka_2.13-2.4.0，两者都与Debian 10.1一起使用。

我首先使用融合版本5.4来鼓励这种行为，认为这可能是其程序包中的一个错误，因为它们具有复制器，并且不应该真正在乎mm2，但是我直接从Apache再现了与kafka_2.13-2.4.0完全相同的问题，改变。

我知道mm2还不是幂等的，不能保证一次交货。在我的测试中(我尝试了很多事情，包括生产者调整或大批数千条消息)。在所有这些测试中，mm2始终复制X3的所有消息。

我错过了什么吗，有人鼓励过同样的事情吗？作为具有相同软件包的旧版mm1的站点说明，我没有这个问题。

感谢任何帮助...谢谢!

即使变更日志没有让我对改进感到非常有信心，我还是再次尝试从kafka 2.4.1运行mm2。 =>这些奇怪的重复并没有改变。

我在新服务器上安装了此发行版，以确保遇到的奇怪行为不是与该服务器相关的事情。

在使用ACL时，我需要特殊的权利吗？我把“所有”都认为是不可能渗透的……即使mm2不是幂等的，我也会尝试与此有关的右边。

让我感到惊讶的是，我找不到任何报告这样的问题的信息，可以肯定我必须做错了什么，但这是问题所在...

Answer 1

您需要从配置中删除connector.class = org.apache.kafka.connect.mirror.MirrorSourceConnector，因为这是告诉Mirror Maker将此类用于其生成的Heartbeats和Checkpoints连接器以及用于复制数据的Source连接器，并且该类使它们的行为与Source连接器完全相同，因此为什么每次都复制3条消息，却实际上生成了3个Source连接器。