Currently I am trying out a Kafka (3.7 version) deployment 1*1 [One Active
(say DC1 ) and One Passive(say DC2)] . And using MIrrormaker-2 for
replicating Kafka data between Active and Passive Data Centers. Here Kafka
and Mirrormaker are executing in Docker Container and volume mounting only
with the log directory of containers with Host VM. Also MM2 is starting on
Passive DC. Replication is happening properly without any problem. After
switchover i.e. when DC2 becomes Active and DC1 becomes Passive,
replication is also working fine but Mirrormaker is duplicating all Kafka
records from scratch from DC2 → DC1. I had to delete all the data on DC1,
to avoid duplicate records.
Question: Is there any way to tell Mirrormaker to start the replication
from a point instead of starting to avoid duplicates. As we are deleting
the data on new Standby and have to start the replication from scratch,
it's taking a lot of time.
Approaches I have tried:
1.
Consumer auto offset latest: With Kafka 3.6.2 and 3.7.0 we have a
property "{{SOURCE_DC_CENTER}}.consumer.auto.offset.reset = latest" as
per Jira: https://issues.apache.org/jira/browse/KAFKA-13988. I tried
this property. I might be wrong, but felt this is a risky one, as in few
scenarios data replication might be missing, especially whenever there is a
delay in Mirrormaker startup and source kafka starts before it and pushes
the data. And also some time if MM stops in between because of some reason,
even then data might be missing to replicate. Questions: Am I right in my
thoughts? Is there any way we can use it well without any issues?
2. Configure first class offset for Mirrormaker Connector: Though we can
set offset here for the Mirrormaker connector using REST API, so that
Mirrormaker Connector can replicate from that point in the new standby
using Rest API.
- As per KIP-710
<https://cwiki.apache.org/confluence/display/KAFKA/KIP-710%3A+Full+support+for+distributed+mode+in+dedicated+MirrorMaker+2.0+clusters>
MM
can expose rest API for Connectors
- As per KIP-875
<https://cwiki.apache.org/confluence/display/KAFKA/KIP-875%3A+First-class+offsets+support+in+Kafka+Connect>
one
can set offsets in Kafka Connector
- Rest server is started, but for all URLs 404 (ex.:
localhost:8083/connectors , response = {404, Not Found})
Questions: Am I missing anything here? Is there any way we can use it
well without any issues?
The following are the config files:
Before Switchover:Active DC1 Kafka server.properties
# This file contains configurations for the Kafka broker -
https://kafka.apache.org/documentation/#configuration
############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each
broker.
broker.id=1
# Allow auto-creation of a non-existent topic on consumer subscribe &
producer assignment
auto.create.topics.enable=false
# Enable topic deletion
delete.topic.enable=false
# Replication configs - number of copies of data across multiple brokers
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
############################# Log Basics #############################
# The directory under which to store log files
log.dir=/opt/kafka/logs
############################# Zookeeper #############################
# Zk connection string (see zk docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=zookeeper:2181
zookeeper.connection.timeout.ms=10000
zookeeper.session.timeout.ms=10000
############################# Additional Broker Settings
#######################
# The maximum allowed session timeout for registered consumers
group.max.session.timeout.ms=300000
# Enable authorization, allow clients connected through non-authenticated
# interfaces (e.g. internal clients) to retain their previous rights.
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:ANONYMOUS
listeners=PLAINTEXT://192.168.1.5:9092
advertised.listeners=PLAINTEXT://192.168.1.5:9092
Passive DC2 Kafka server.properties
# This file contains configurations for the Kafka broker -
https://kafka.apache.org/documentation/#configuration
############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each
broker.
broker.id=2
# Allow auto-creation of a non-existent topic on consumer subscribe &
producer assignment
auto.create.topics.enable=false
# Enable topic deletion
delete.topic.enable=false
# Replication configs - number of copies of data across multiple brokers
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
############################# Log Basics #############################
# The directory under which to store log files
log.dir=/opt/kafka/logs
############################# Zookeeper #############################
# Zk connection string (see zk docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=zookeeper:2181
zookeeper.connection.timeout.ms=10000
zookeeper.session.timeout.ms=10000
############################# Additional Broker Settings
#######################
# The maximum allowed session timeout for registered consumers
group.max.session.timeout.ms=300000
# Enable authorization, allow clients connected through non-authenticated
# interfaces (e.g. internal clients) to retain their previous rights.
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:ANONYMOUS
listeners=PLAINTEXT://192.168.2.5:9092
advertised.listeners=PLAINTEXT://192.168.2.5:9092
Passive DC2 MM2 connect-mirrormaker.properties
# 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=DC1, DC2
# 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"
DC1.bootstrap.servers=kafka.node1:9092
DC2.bootstrap.servers=kafka.node2:9092
# enable and configure individual replication flows
DC1->DC2.enabled=true
# regex which defines which topics gets replicated. For eg "foo-.*"
DC1->DC2.topics=.*
# Setting replication factor of newly created remote topics
replication.factor=1
############################# 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.
checkpoints.topic.replication.factor=1
heartbeats.topic.replication.factor=1
offset-syncs.topic.replication.factor=1
# 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.
offset.storage.replication.factor=1
status.storage.replication.factor=1
config.storage.replication.factor=1
replication.policy.class=org.apache.kafka.connect.mirror.IdentityReplicationPolicy
DC1->DC2.sync.group.offsets.enabled=false
DC1->DC2.sync.group.offsets.interval.seconds = 5
DC1->DC2.refresh.topics.enabled = true
DC1->DC2.refresh.groups.enabled = true
DC1->DC2.emit.checkpoints.enabled = true
DC1->DC2.emit.heartbeats.enabled = true
DC1->DC2.sync.topic.configs.enabled = true
DC1->DC2.sync.topic.acls.enabled = true
DC2->DC1.enabled = false
DC2->DC1.emit.heartbeats.enabled = false
DC1->DC2.refresh.topics.interval.seconds=5
DC1->DC2.refresh.groups.interval.seconds = 5
DC1->DC2.sync.topic.acls.interval.seconds = 5
DC1->DC2.sync.topic.configs.interval.seconds = 5
dedicated.mode.enable.internal.rest=true
After Switchover:Active DC2 Kafka server.properties
# This file contains configurations for the Kafka broker -
https://kafka.apache.org/documentation/#configuration
############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each
broker.
broker.id=2
# Allow auto-creation of a non-existent topic on consumer subscribe &
producer assignment
auto.create.topics.enable=false
# Enable topic deletion
delete.topic.enable=false
# Replication configs - number of copies of data across multiple brokers
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
############################# Log Basics #############################
# The directory under which to store log files
log.dir=/opt/kafka/logs
############################# Zookeeper #############################
# Zk connection string (see zk docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=zookeeper:2181
zookeeper.connection.timeout.ms=10000
zookeeper.session.timeout.ms=10000
############################# Additional Broker Settings
#######################
# The maximum allowed session timeout for registered consumers
group.max.session.timeout.ms=300000
# Enable authorization, allow clients connected through non-authenticated
# interfaces (e.g. internal clients) to retain their previous rights.
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:ANONYMOUS
listeners=PLAINTEXT://192.168.2.5:9092
advertised.listeners=PLAINTEXT://192.168.2.5:9092
Passive DC1 Kafka server.properties
# This file contains configurations for the Kafka broker -
https://kafka.apache.org/documentation/#configuration
############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each
broker.
broker.id=1
# Allow auto-creation of a non-existent topic on consumer subscribe &
producer assignment
auto.create.topics.enable=false
# Enable topic deletion
delete.topic.enable=false
# Replication configs - number of copies of data across multiple brokers
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
############################# Log Basics #############################
# The directory under which to store log files
log.dir=/opt/kafka/logs
############################# Zookeeper #############################
# Zk connection string (see zk docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=zookeeper:2181
zookeeper.connection.timeout.ms=10000
zookeeper.session.timeout.ms=10000
############################# Additional Broker Settings
#######################
# The maximum allowed session timeout for registered consumers
group.max.session.timeout.ms=300000
# Enable authorization, allow clients connected through non-authenticated
# interfaces (e.g. internal clients) to retain their previous rights.
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:ANONYMOUS
listeners=PLAINTEXT://192.168.1.5:9092
advertised.listeners=PLAINTEXT://192.168.1.5:9092
Passive DC1 MM2 connect-mirrormaker.properties
# 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=DC2, DC1
# 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"
DC2.bootstrap.servers=kafka.node2:9092
DC1.bootstrap.servers=kafka.node1:9092
# enable and configure individual replication flows
DC2->DC1.enabled=true
# regex which defines which topics gets replicated. For eg "foo-.*"
DC2->DC1.topics=.*
# Setting replication factor of newly created remote topics
replication.factor=1
############################# 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.
checkpoints.topic.replication.factor=1
heartbeats.topic.replication.factor=1
offset-syncs.topic.replication.factor=1
# 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.
offset.storage.replication.factor=1
status.storage.replication.factor=1
config.storage.replication.factor=1
replication.policy.class=org.apache.kafka.connect.mirror.IdentityReplicationPolicy
DC2->DC1.sync.group.offsets.enabled=false
DC2->DC1.sync.group.offsets.interval.seconds = 5
DC2->DC1.refresh.topics.enabled = true
DC2->DC1.refresh.groups.enabled = true
DC2->DC1.emit.checkpoints.enabled = true
DC2->DC1.emit.heartbeats.enabled = true
DC2->DC1.sync.topic.configs.enabled = true
DC2->DC1.sync.topic.acls.enabled = true
DC1->DC2.enabled = false
DC1->DC2.emit.heartbeats.enabled = false
DC2->DC1.refresh.topics.interval.seconds=5
DC2->DC1.refresh.groups.interval.seconds = 5
DC2->DC1.sync.topic.acls.interval.seconds = 5
DC2->DC1.sync.topic.configs.interval.seconds = 5
dedicated.mode.enable.internal.rest=true
Please let me know if additional information is needed.
Thanks,
Vijay
