kafka异步复制配置,kafka跨集群同步消息配置
kafka异步复制配置,kafka跨集群同步消息配置,connect-mirror-maker.properties
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
# Licensed to the Apache Software Foundation (ASF) under A or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # see org.apache.kafka.clients.consumer.ConsumerConfig for more details # Sample MirrorMaker 2.0 top-level configuration file # Run with ./bin/connect-mirror-maker.sh connect-mirror-maker.properties # specify any number of cluster aliases # 指定两个集群 clusters = A, B # A集群认证,如需要 #A.bootstrap.servers=CLUSTER-NAME-kafka-bootstrap-PROJECT-NAME:443 #A.security.protocol=SSL #A.ssl.truststore.password=TRUSTSTORE-NAME #A.ssl.truststore.location=PATH-TO-TRUSTSTORE/truststore.cluster-1.jks #A.ssl.keystore.password=KEYSTORE-NAME #A.ssl.keystore.location=PATH-TO-KEYSTORE/user.cluster-1.p12_ # B集群认证,如需要 #B.bootstrap.servers=CLUSTER-NAME-kafka-bootstrap-<my-project>:443 #B.security.protocol=SSL #B.ssl.truststore.password=TRUSTSTORE-NAME #B.ssl.truststore.location=PATH-TO-TRUSTSTORE/truststore.cluster-2.jks_ #B.ssl.keystore.password=KEYSTORE-NAME #B.ssl.keystore.location=PATH-TO-KEYSTORE/user.cluster-2.p12_ # connection information for each cluster # This is a comma separated host:port pairs for each cluster # for e.g. "A_host1:9092, A_host2:9092, A_host3:9092" A.bootstrap.servers = 52.77.248.99:32768, 52.77.248.99:32769, 52.77.248.99:32770 B.bootstrap.servers = 54.255.46.134:32768, 54.255.46.134:32769, 54.255.46.134:32770 # enable and configure individual replication flows # 指定数据同步方向,可以单向同步也可互相同步 A->B.enabled = true # regex which defines which topics gets replicated. For eg "foo-.*" # 指定同步的Topic,支持正则匹配,默认复制全部Topic,如:"foo-.*" # 全部同步 A->B.topics = .* # 设置同步的topic;支持正则 #A->B.topics = xxxx,xxxx # 设置排除的topic:支持正则 #A->B.topics.exclude= xxxx # 打开以下两个配置则表示A、B两个集群互相复制同步 B->A.enabled = true B->A.topics = .* # Setting replication factor of newly created remote topics # 设置副本个数,如果是要同步多个Topic且副本数各不相同,建议先创建同名同本数的Topic再启动MirrorMaker replication.factor=3 ############################# Internal Topic Settings ############################# # The replication factor for mm2 internal topics "heartbeats", "B.checkpoints.internal" and # "mm2-offset-syncs.B.internal" # For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3. sync.topic.configs.enabled=true #同步配置的时间频率 sync.topic.configs.enabled.interval.seconds=60 # 测试环境可以为1,生产环境建议以下配置大于1,比如设为3 checkpoints.topic.replication.factor=3 heartbeats.topic.replication.factor=3 offset-syncs.topic.replication.factor=3 # The replication factor for connect internal topics "mm2-configs.B.internal", "mm2-offsets.B.internal" and # "mm2-status.B.internal" # For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3. # 测试环境可以为1,生产环境建议以下配置大于1,比如设为3 offset.storage.replication.factor=3 status.storage.replication.factor=3 config.storage.replication.factor=3 # customize as needed # replication.policy.separator = _ sync.topic.acls.enabled = true # 心跳频率 emit.heartbeats.interval.seconds = 10 # 开启topic动态和消费者组 动态同步与同步的周期 refresh.topics.enabled = true # 检查源集群中是否有新主题的频率 refresh.topics.interval.seconds=60 refresh.groups.enabled = true # 检查源集群中要同步的新消费组的频率 refresh.groups.interval.seconds=60 # 开始消费者组offset同步;设置同步的周期---注意:仅仅同步idle中的消费者的offset sync.group.offsets.enabled = true sync.group.offsets.interval.seconds = 5 # 启动同步的Task数量----启用几个线程进行同步 tasks.max = 20 # 设置同步的topic Name命名规则;3.0版本提供了两种topic同步命名规则,默认会带上前缀,也可以手动不带前缀的----此时不能做双向同步 # 打开此选项则会去掉同步的前缀 # replication.policy.class = org.apache.kafka.connect.mirror.IdentityReplicationPolicy # 启动命令,最好在目标集群机器上启动。 # bin/connect-mirror-maker.sh config/connect-mirror-maker.properties --clusters B |