Kafka MirrorMaker2 主题 - 复制因子

Question

我正在本地测试 kafka mirror maker 2。运行 2 个动物园管理员和 2 个 kafka 经纪人。它们作为单个实例运行。因此，我本地有 2 个集群，每个集群上有 1 个代理。经纪人运作良好。现在，当我尝试在本地运行镜像制造商时，遇到以下问题。

[2020-02-14 16:58:18,371] ERROR [Worker clientId=connect-2, groupId=dest-mm2] Uncaught exception in herder work thread, exiting:  (org.apache.kafka.connect.runtime.distributed.DistributedHerder:290)
org.apache.kafka.connect.errors.ConnectException: Error while attempting to create/find topic(s) 'mm2-offsets.dest.internal'
    at org.apache.kafka.connect.util.TopicAdmin.createTopics(TopicAdmin.java:262)
    at org.apache.kafka.connect.storage.KafkaOffsetBackingStore$1.run(KafkaOffsetBackingStore.java:99)
    at org.apache.kafka.connect.util.KafkaBasedLog.start(KafkaBasedLog.java:128)
    at org.apache.kafka.connect.storage.KafkaOffsetBackingStore.start(KafkaOffsetBackingStore.java:109)
    at org.apache.kafka.connect.runtime.Worker.start(Worker.java:186)
    at org.apache.kafka.connect.runtime.AbstractHerder.startServices(AbstractHerder.java:121)
    at org.apache.kafka.connect.runtime.distributed.DistributedHerder.run(DistributedHerder.java:277)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 3 larger than available brokers: 1.
    at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
    at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
    at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
    at org.apache.kafka.connect.util.TopicAdmin.createTopics(TopicAdmin.java:229)
    ... 11 more
Caused by: org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 3 larger than available brokers: 1.
[2020-02-14 16:58:18,371] ERROR [Worker clientId=connect-1, groupId=source-mm2] Uncaught exception in herder work thread, exiting:  (org.apache.kafka.connect.runtime.distributed.DistributedHerder:290)
org.apache.kafka.connect.errors.ConnectException: Error while attempting to create/find topic(s) 'mm2-offsets.source.internal'
    at org.apache.kafka.connect.util.TopicAdmin.createTopics(TopicAdmin.java:262)
    at org.apache.kafka.connect.storage.KafkaOffsetBackingStore$1.run(KafkaOffsetBackingStore.java:99)
    at org.apache.kafka.connect.util.KafkaBasedLog.start(KafkaBasedLog.java:128)
    at org.apache.kafka.connect.storage.KafkaOffsetBackingStore.start(KafkaOffsetBackingStore.java:109)
    at org.apache.kafka.connect.runtime.Worker.start(Worker.java:186)
    at org.apache.kafka.connect.runtime.AbstractHerder.startServices(AbstractHerder.java:121)
    at org.apache.kafka.connect.runtime.distributed.DistributedHerder.run(DistributedHerder.java:277)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 3 larger than available brokers: 1.
    at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
    at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
    at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
    at org.apache.kafka.connect.util.TopicAdmin.createTopics(TopicAdmin.java:229)
    ... 11 more
Caused by: org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 3 larger than available brokers: 1.

发生此错误是因为我的本地和镜像制造商中只有一个代理正在尝试创建复制因子为 3 的偏移主题。如何设置配置以将复制因子更改为 1。

我尝试如下，但仍然是同样的错误。不知道什么是正确的配置。

mm2.properties

# mm2.properties 
name=source->dest 
clusters=source, dest

source.bootstrap.servers=localhost:9091 
dest.bootstrap.servers=localhost:9092

source->dest.enabled=true

topics=test-3 groups=test-6-group 
offsets.storage.replication.factor=1
source.replication.factor=1 
dest.replication.factor=1

mm2-offsets.source.internal.replication.factor=1 
mm2-offsets.dest.internal.replication.factor=1

Answer 1

添加以下 2 个道具后即可使用。

config.storage.replication.factor=1
status.storage.replication.factor=1

Answer 2

设置为'offset'（单数）

用途：

offset.storage.replication.factor=1

代替：

offsets.storage.replication.factor=1

...帮了我一把