2.1 Advantages of using the Kafka Connect framework¶

• Kafka Connect offers dynamic configuration via a REST API and has a framework for distributing work across the Kafka cluster when more throughput is needed.

• We are already using Kafka Connect to run the InfluxDB Sink, Amazon S3 Sink and the JDBC source and sink connectors.

• We are familiar with the Kafka Connect management API, and we can reuse some of the tools we already have in place.

• We are also familiar with the Kafka Connect metrics for monitoring.

2.2 Replicating topics and schemas¶

In the EFD, the creation of new kafka topics in the source cluster is done by SAL Kafka. That happens dynamically as SAL Kafka subscribes to new CSCs. In the destination cluster, that must be done by the replication solution. That means the replication solution must be able to automatically detect new topics and changes in topic configuration in the source cluster and propagate them to the destination cluster.

We use Avro to encode records in Kafka. SAL Kafka is also responsible for creating the Avro schemas and register them with the Schema registry in the source cluster. The replication solution also must be able to replicate the schemas to the destination cluster.

2.3 Active/standby and active/active replication scenarios¶

In the active/standby replication scenario, data replication happens in only one direction, from the active Kafka cluster (Summit EFD) to the standby Kafka cluster (LDF EFD). The active/standby scenario works well for the EFD use case. In the EFD use case, there is no need to modify the data at LDF so we don’t need to replicate it back to the Summit. An important simplification of the active/standby scenario is that we can preserve topic names in the source and destination Kafka clusters. By preserving the topic names, we can execute the same queries, dashboards and notebooks at the Summit EFD and at the LDF EFD transparently.1 As discussed in section 4.4   Migrating Avro schemas, in the active/standby scenario we can migrate Avro schemas without modification.

The active/active scenario may be important for other applications like the OWL system. In the OWL use case, the data can be modified at LDF and therefore it must be replicated back to the Summit. The active/active replication scenario requires topic renaming and consequently schema translation when synchronizing the topic schemas between the Kafka clusters.

1

If topics are renamed in Kafka, to achieve the same result we would have to rename them back to their original names in each connector individually when writing to other formats.

3.1 Brooklin¶

Brooklin is developed at LinkedIn and was designed as a generic bridge for streaming data. It is comparable to solutions in the streaming ecosystem like Facebook’s Wormhole and Kafka Connect. You can learn more about Brooklin from their engineering blog post Streaming Data Pipelines with Brooklin

It has successfully replaced MirrorMaker at Linkedin (see also the Waifair use case). It does more than we need and it is not based on the Kafka connect framework. Adopting Brooklin would add unnecessary complexity to the current EFD architecture.

3.2 Uber uReplicator¶

We could build on Uber’s uReplicator, which solves some of MirrorMaker problems. However, uReplicator uses Apache Helix to provide some of the features already present in Kafka Connect, like a REST API, dynamic configuration changes, cluster management, coordination, etc. We choose to give preference to solutions that use native components of Apache Kafka. That said, uReplicator was also a major source of inspiration for the MirrorMaker 2 design.

3.3 Mirus¶

Mirus is developed at Salesforce. It is essentially a custom Kafka Connect Source Connector specialized for reading data from multiple Kafka clusters. Mirus also includes a monitor thread that looks for task and connector failures with optional auto-restarts, and custom JMX metrics for monitoring. You can read more about Mirus in this blog post.

Mirus is able to replicate new topics partitions created in the source cluster to the destination cluster, however it cannot replicate other topic configurations.

3.4 MirrorMaker 2¶

KIP-382 proposed MirrorMaker 2 as a replacement for MirrorMaker with a new multi-cluster, cross-data-center replication engine based on the Kafka Connect framework, recognizing the need for a native replication solution for Apache Kafka.

Both Confluent Replicator and MirrorMaker2 leverage the Kafka Connect framework and share many design principles. One feature that is key for full replication is to preserve identical topic configuration between source and destination cluster. For example, if new partitions are added on to the source cluster, that should be automatically detected and replicated to the destination cluster.

MirrorMaker 2 introduces the concept of remote topics, which are replicated topics referencing a source topic via a naming convention. Any partitions in a remote topic are remote partitions and refer to the same partitions in the source topic.

Because MirrorMaker 2 is the native replication solution for Apache Kafka and is built on the Kafka Connect framework it is the best alternative to replace Confluent Replicator for the EFD replication service.

4.1 Configuring the connectors¶

docker-compose run --entrypoint /bin/bash kafkaconnect

cat <<EOF > heartbeat-connector.json
{
  "name": "heartbeat",
  "connector.class": "org.apache.kafka.connect.mirror.MirrorHeartbeatConnector",
  "source.cluster.alias": "src",
  "target.cluster.alias": "destn",
  "source.cluster.bootstrap.servers": "broker-src:29093",
  "topics": "foo"
}
EOF
kafkaconnect upload --name heartbeat heartbeat-connector.json

kafkaconnect status heartbeat

cat <<EOF > checkpoint-connector.json
{
  "connector.class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
  "name": "checkpoint",
  "source.cluster.alias": "src",
  "source.cluster.bootstrap.servers": "broker-src:29093",
  "target.cluster.alias": "destn",
  "target.cluster.bootstrap.servers": "broker-destn:29092",
  "topics": "foo"
}
EOF
kafkaconnect upload --name checkpoint checkpoint-connector.json

cat <<EOF > mirror-source-connector.json
{
  "checkpoints.topic.replication.factor": 1,
  "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
  "key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
  "heartbeats.topic.replication.factor": 1,
  "name": "mirror-source",
  "offset-syncs.topic.replication.factor": 1,
  "refresh.topics.interval.seconds": 60,
  "replication.factor": 1,
  "source.cluster.alias": "src",
  "source.cluster.bootstrap.servers": "broker-src:29093",
  "sync.topic.acls.enabled": false,
  "sync.topic.configs.enabled": true,
  "sync.topic.configs.interval.seconds": 60,
  "target.cluster.alias": "destn",
  "target.cluster.bootstrap.servers": "broker-destn:29092",
  "tasks.max": "1",
  "topics": "foo",
  "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter"
}
EOF
kafkaconnect upload --name mirror-source mirror-source-connector.json

docker-compose logs connect-destn

4.2 Creating a topic in the source cluster¶

Lets create the topic foo with one partition in the source cluster:

docker-compose exec broker-src kafka-topics --bootstrap-server broker-src:29093 --create --topic foo --partitions 1 --replication-factor 1

The remote topic is then replicated to the destination cluster:

docker-compose exec broker-destn kafka-topics --bootstrap-server broker-destn:29092 --list

Notice that, by default, the remote topic name is prefixed by the source cluster alias src.foo.

4.3 Producing Avro records¶

You can use the kafka-avro-console-producer tool to produce Avro records to the foo topic in the source cluster. kafka-avro-console-producer can also register the Avro schema for the foo topic with the Schema Registry. In the example, we create a simple schema for the foo topic with the fields baz and bar.

docker-compose exec schema-registry-src kafka-avro-console-producer --bootstrap-server broker-src:29093 --topic foo --property value.schema='{"type": "record", "name": "foo", "fields": [{"name": "bar", "type": "string"}, {"name": "baz", "type": "float"}]}'

You can paste a few records and then finish with Ctrl+D.

{"bar": "John Doe", "baz": 1}
{"bar": "John Doe", "baz": 1}
{"bar": "John Doe", "baz": 1}

From the Kafka Connect logs you can see MirrorSource pulling records from the foo topic and committing offsets for its consumer.

Finally, you can verify that a change in the original topic configuration is in fact replicated to the remote topic. For example, this command will add another partition to the foo topic:

docker-compose exec broker-src kafka-topics --bootstrap-server broker-src:29093 --alter --topic foo --partitions 2

and the new partition is added to the remote topic after 60 seconds:

docker-compose exec broker-destn kafka-topics --bootstrap-server broker-destn:29092 --describe --topic src.foo

4.4 Migrating Avro schemas¶

Avro schemas are needed at the destination cluster to deserialize the Avro messages before writing to InfluxDB or to other storage formats.

To replicate schemas, we need to replicate the _schemas internal topic to the destination cluster. That is done by explicitly adding this topic to the topics configuration property in the MirrorHeartbeat, MirrorCheckpoint and MirrorSource connector configurations.

One problem, however, is that, by default, remote topics are renamed, and Kafka Connect will fail trying to find a subject name that corresponds to the default remote topic name.

Confluent Replicator solves this problem with schema translation (see 7   Appendix A - Confluent Replicator). That feature is missing in MirrorMaker 2.

A possible solution in MirrorMaker 2 is to set "source.cluster.alias": "" and "replication.policy.separator": "". This way because topics and subject names are preserved, no schema translation is required. Aslo the name of the _schemas topic is preserved no specific configuration is required for the Schema Registry at the destination cluster.

Once you apply these configurations, you can use the kafka-avro-console-consumer tool to consume records from the remote topic foo using the Schema Registry at the destination cluster to deserialize the messages:

docker-compose exec schema-registry-destn kafka-avro-console-consumer --topic foo --from-beginning --bootstrap-server  broker-destn:29092 --property schema.registry.url=http://schema-registry-destn:8084