KafkaMirrorMaker2.0架构与实战全解《二》

* As a dedicated MirrorMaker cluster.
* As a Connector in a distributed Connect cluster.
* As a standalone Connect worker.

#does not require an existing Connect cluster. Instead, a high-level driver manages a collection of Connect workers.
$ ./bin/connect-mirror-maker.sh mm2.properties

#In this mode, a single Connect worker runs MirrorSourceConnector. 
#This does not support multinode clusters, but is useful for small workloads or for testing.
$ ./bin/connecti-standalone.sh worker.properties connector.properties

#If you already have a Connect cluster, you can configure it to run MirrorMaker connectors.
#There are four such connectors:MirrorSourceConnector、MirrorSinkConnector、MirrorCheckpointConnector、MirrorHeartbeatConnector
Configure these using the Connect REST API:
PUT connectors/us-west-source/config HTTP/1.1
{
    "name": "us-west-source",
    "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "source.cluster.alias": "us-west",
    "target.cluster.alias": "us-east",
    "source.cluster.bootstrap.servers": "us-west-host1:9091",
    "topics": ".*"
}

Connectors：the high level abstraction that coordinates data streaming by managing tasks
Tasks：the implementation of how data is copied to or from Kafka
Workers：the running processes that execute connectors and tasks

# 集群级别 
clusters = us-west, us-east   # defines cluster aliases
us-west.bootstrap.servers = broker3-west:9092
us-east.bootstrap.servers = broker5-east:9092
topics = .*   # all topics to be replicated by default
topics.exclude = .*[\-\.]internal, .*\.replica, __.*
groups = .*
groups.exclude = console-consumer-.*, connect-.*, __.*
##检查新建topics
refresh.topics.enabled = true
refresh.topics.interval.seconds = 600
##检查新建groups
refresh.groups.enabled = true
refresh.groups.interval.seconds = 600
##同步topic config
sync.topic.configs.enabled = true
##同步acl
sync.topic.acls.enabled = true
##心跳监测
emit.heartbeats.enabled = true
emit.heartbeats.interval.seconds = 1
heartbeats.topic.replication.factor = 3
##checkpoint设置
emit.checkpoints.enabled = true
emit.checkpoints.interval.seconds = 60
checkpoints.topic.replication.factor = 3
##同步group offset
sync.group.offsets.enabled = true 
sync.group.offsets.interval.seconds = 60
offset-syncs.topic.replication.factor = 3


# 数据流级别
us-west->us-east.enabled = true         # flow from us-west to us-east
us-west->us.east.topics = foo.*, bar.*  # override the default above
us-west->us.east.groups = foo.*, bar.*  # override the default above
##定制target topic分隔符
us-west->us-east.replication.policy.separator = _  #default(.)
##如果需要定制特殊映射关系，可以参考DefaultReplicationPolicy类实现ReplicationPolicy和Configurable两个接口，详见源码
##https://github.com/apache/kafka/blob/trunk/connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/DefaultReplicationPolicy.java

# 框架级别参数
# Setting Kafka Connect defaults for MirrorMaker
tasks.max = 5  # 默认框架参数开箱即用(out-of-the-box)，唯独task.max需要依情况设置(建议至少为2，需考虑硬件资源、复制topic-partitions数目)


# connector work-level级别参数
# Use the format of {cluster}.{config_name}
us-west.offset.storage.topic = my-mirrormaker-offsets

* {source}.consumer.{consumer_config_name}
* {target}.producer.{producer_config_name}
* {source_or_target}.admin.{admin_config_name}

consumer举例：us-west cluster (from which to consume)
us-west.consumer.isolation.level = read_committed
us-west.admin.bootstrap.servers = broker57-primary:9092


producer举例：us-east cluster (to which to produce)
us-east.producer.compression.type = gzip
us-east.producer.buffer.memory = 32768


admin举例：
us-east.admin.bootstrap.servers = broker8-secondary:9092

# MBean: kafka.connect.mirror:type=MirrorSourceConnector,target=([-.w]+),topic=([-.w]+),partition=([0-9]+)
record-count            # number of records replicated source -> target
record-age-ms           # age of records when they are replicated
record-age-ms-min
record-age-ms-max
record-age-ms-avg
replication-latency-ms  # time it takes records to propagate source->target
replication-latency-ms-min
replication-latency-ms-max
replication-latency-ms-avg
byte-rate               # average number of bytes/sec in replicated records
# MBean: kafka.connect.mirror:type=MirrorCheckpointConnector,source=([-.w]+),target=([-.w]+)
checkpoint-latency-ms   # time it takes to replicate consumer offsets
checkpoint-latency-ms-min
checkpoint-latency-ms-max
checkpoint-latency-ms-avg