Kafka 4 MirrorMaker2

Kafka MirrorMaker2 is integrated with ODP Ambari, and replicates Kafka topics between ODP clusters, allowing data transfer from source to destination and vice versa. Before you configure the Kafka MirrorMaker2 service through Ambari, confirm the health of the Kafka service, including its ability to add and fetch data from topics.

This page describes how to enable Kerberos, SSL, and Ranger authorization for secure access to Kafka service topics, and how to configure and test Kafka4 MirrorMaker2.

Kafka MirrorMaker2 on Kafka4 refers to the same MirrorMaker2 on Kafka3, since Kafka3 already supports the MirrorMaker2 architecture.

Enable Kafka Kerberos

Enabling Kerberos for the entire ODP cluster also enables Kerberos for Kafka. For instructions on enabling Kerberos on an ODP cluster, see Enabling Kerberos in an ODP Cluster.

Enable Kafka SSL

Gather the CA-signed truststore and keystore certificates, along with their details, to enable SSL for the Kafka service. If CA certificates aren't accessible, generate self-signed certificates using the steps below.

Replicate the following script on all nodes where Kafka services are running, and run it. Then copy the $(hostname).crt file to all other nodes reciprocally, and import the certificate into the truststore on each node.

cat Kafka-ssl.sh

password="<password>"
mkdir -p /opt/security/pki/
cd /opt/security/pki/ || exit
# Generate SSL certificate
keytool -genkey -alias "$(hostname)" -keyalg RSA -keysize 2048 -dname \
 "CN=$(hostname -f),OU=SU,O=ACCELO,L=BNG,ST=KN,C=IN" -keypass \
 "$password" -keystore keystore.jks -storepass "$password"
# Export SSL certificate
keytool -export -alias "$(hostname)" -keystore keystore.jks -file \
 "$(hostname).crt" -storepass "$password"
# Import SSL certificate into truststore
yes | keytool -import -file "$(hostname).crt" -keystore truststore.jks \
 -alias "$(hostname)-trust" -storepass "$password"

With the certificates generated above (or your CA certificates), you're ready to configure Kerberos, SSL, and Ranger for Kafka.

Configure Kerberos, SSL, and Ranger for Kafka

Once you've obtained the Kerberos and SSL details described in the preceding sections, configure them for the Kafka service:

listeners=SASL_SSL://localhost:7001,SASL_PLAINTEXT://localhost:7004
authorizer.class.name=org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuthorizer
sasl.enabled.mechanisms=GSSAPI
sasl.mechanism.inter.broker.protocol=GSSAPI
security.inter.broker.protocol=SASL_PLAINTEXT
ssl.keystore.location=/opt/security/pki/keystore.jks
ssl.keystore.password=<password>
ssl.truststore.location=/opt/security/pki/truststore.jks
ssl.truststore.password=<password>
ssl.key.password=<password>
ssl.client.auth=none

Leave all remaining configurations at their default values. Kafka is now enabled with Ranger, Kerberos, and SSL.

Set up Kafka4 MirrorMaker2 with Kerberos and SSL

MirrorMaker2 supports multiple topologies. This page covers one topology, consisting of one source cluster and one destination cluster.

Step 1: Navigate to the Kafka MirrorMaker2 configuration

Sign in to the Ambari UI.
In the main menu, go to the Services list, and select Kafka from the list of services to open the Kafka service page.
On the Kafka service page, select the Configs tab.
Go to the Advanced section, and find the subsection titled Advanced kafka-mirrormaker2.
Modify and save the configuration at Mirror-Maker2 Template.

Step 2: Configure the Mirror-Maker2 Template

clusters=source, dest
 
source->dest.enabled=true
source->dest.topics=.*
source.bootstrap.servers=basa-mm2-01.acceldata.ce:7001,basa-mm2-02.acceldata.ce:7001,basa-mm2-03.acceldata.ce:7001
source.java.security.auth.login.config=/usr/odp/current/kafka4-broker/config/kafka_jaas.conf
source.java.security.krb5.kdc=basa-mm2-01.acceldata.ce:7001
source.java.security.krb5.realm=ADSRE.COM
source.javax.security.auth.useSubjectCredsOnly=true
source.replication.enabled=true
source.sasl.enabled.mechanisms=GSSAPI
source.sasl.kerberos.service.name=kafka
source.security.protocol=SASL_PLAINTEXT
source.ssl.key.password=
source.ssl.keystore.location=
source.ssl.keystore.password=
source.ssl.truststore.location=
source.ssl.truststore.password=
source.sun.security.krb5.debug=true
source.topics.whitelist=.*
 
dest->source.enabled=false
dest->source.topics=.*
dest.bootstrap.servers=mm2cl03.acceldata.ce:6667
dest.java.security.auth.login.config=/usr/odp/current/kafka-broker/config/kafka_jaas.conf
dest.java.security.krb5.kdc=
dest.java.security.krb5.realm=
dest.javax.security.auth.useSubjectCredsOnly=
dest.replication.enabled=false
dest.sasl.enabled.mechanisms=GSSAPI
dest.sasl.kerberos.service.name=kafka
dest.security.protocol=PLAINTEXT
dest.ssl.key.password=
dest.ssl.keystore.location=
dest.ssl.keystore.password=
dest.ssl.truststore.location=
dest.ssl.truststore.password=
dest.sun.security.krb5.debug=true
dest.topics.whitelist=.*
 
refresh.topics.enabled=true
refresh.topics.interval.seconds=5
replication.factor=1
status.storage.replication.factor=1
heartbeats.topic.replication.factor=1
offset-syncs.topic.replication.factor=1
offset.storage.replication.factor=1
config.storage.replication.factor=1
checkpoints.topic.replication.factor=1

This example uses a replication factor of 1 because it's based on a test cluster. If your Kafka clusters consist of 3 or more nodes, set the replication factor to a minimum of 3.

Test the MirrorMaker2 use case

The following topics are replicated from the source cluster to the destination cluster, each appended with .Dest:

./bin/kafka-topics.sh --bootstrap-server mm2cl03.acceldata.ce:7001 --list --command-config client-ssl.properties

__consumer_offsets
conn-test-0104-01
conn-test-0104-02
conn-test-0104-03
connect-configs
connect-offsets
connect-status
heartbeats
mm2-configs.Dest.internal
mm2-offset-syncs.Dest.internal
mm2-offsets.Dest.internal
mm2-status.Dest.internal

In the destination cluster, you can see the following default topics, each appended with the name of the source cluster:

./bin/kafka-topics.sh --bootstrap-server basa-mm2-01.acceldata.ce:7001 --list

__CruiseControlMetrics
__consumer_offsets
connect-configs
connect-offsets
connect-status
customers.students10
customers.students2
heartbeats
mm2-configs.Source.internal
mm2-offsets.Source.internal
mm2-status.Source.internal

Replicate a topic from the source to the destination cluster

Step 1: Produce data on the source cluster

Produce data for your topic using the Kafka console producer. Connect to your source cluster and specify the topic and security configurations:

./bin/kafka-console-producer.sh \
 --bootstrap-server basa-mm2-01.acceldata.ce:7001 \
 --topic sd-1203-10001 \
 --producer.config client-ssl.properties

Add sample data:

>org.apache.kafka.connect.mirror.Scheduler1
>org.apache.kafka.connect.mirror.Scheduler2

Step 2: Verify from the destination cluster

Confirm that the topic replicated to the destination cluster. Use the Kafka topics command to list topics and verify the presence of your source topic:

[root@odp10 kafka4]#
./bin/kafka-topics.sh \
 --bootstrap-server mm2cl03.acceldata.ce:6667 \
 --list | grep Source

Expected output:

Source.sd-1203-10001

Step 3: Consume data from the replicated cluster

Use the Kafka console consumer to view the data from the replicated topic and confirm successful replication:

./bin/kafka-console-consumer.sh \
 --topic Source.sd-1203-10001 \
 --bootstrap-server mm2cl03.acceldata.ce:6667 \
 --from-beginning