MirrorTool for Kafka Connect (Deprecated)

Note: Since version 2.4 Kafka ships with a Kafka Source Connector that implements this functionality. As such, this repository is no longer actively maintained.

A Kafka Source Connector for Kafka Connect. Mirror topics from a source Kafka cluster into your destination Kafka cluster.

Downloads

You can grab a pre-built jar on the Releases page.

Why

The aim of this connector is to enable MirrorMaker-like functionality within the Kafka Connect framework.

At Comcast, we have seen two key benefits to using MirrorTool over MirrorMaker:

• No access to the ZooKeeper on the source kafka clusters required.
• Can use existing Kafka Connect deployment, instead of having to manage and monitor another stand alone application

How it works

• Connector starts up a "Partition Monitor" thread which uses the Kafka AdminClient to get a list of topics (and partitons) from the source cluster to be mirrored to the destination cluster.
• List of available topic partitions are allocated to the configured number of tasks.
• Each task starts a KafkaConsumer and assigns its allocated topic partitions to the consumer.
• As messages are consumed from the source cluster, they are passed to the Kafka Connect framework to deliver the messages to the destination kafka cluster
• The "Partition Monitor" thread periodically checks the source cluster for changes to the topics and partitions that have been configured for the connector, and will trigger a task reconfiguration if required.

Build

Build a plugin jar to build/libs/MirrorTool-[version].jar by running:

./gradlew build

Copy the resulting jar file into your Kafka Connect plugin directory.

Configuration Options

Note that these options are for this MirrorTool connector, and assumes some familiarity with the base Kafka Connect configuration options. (You will need to configure the parameters for your destination Kafka Cluster in the base Kafka Connect configuration)

Common Options

These are the most common options that are required when configuring this connector:

• "Stored offset" here does not mean the stored consumer group offset, rather the stored offset within the Kafka Connect offset.storage.topic topic. If you want to start the connector from an existing consumer group, then set source.auto.offset.reset to none, and update source.group.id accordingly.

Advanced Options

Some use cases may require modifying the following default connector options. Use with care.

Overriding the internal KafkaConsumer and AdminClient Configuration

Note that standard Kafka parameters can be passed to the internal KafkaConsumer and AdminClient by prefixing the standard configuration parameters with "source.". For cases where the configuration for the KafkaConsumer and AdminClient diverges, you can use the more explicit "connector.consumer." and "connector.admin." configuration parameter prefixes to fine tune the settings used for each.

Example Configuration

{
  "name": "kafka-connect-kafka-source-example", // Name of the kafka connect task
    "config": {
    "tasks.max": "2", // Maximum number of parallel tasks to run
    "key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter", // Set the format of data that is written to kafka. As the connector parses source cluster messages as binary arrays, this will be the usual required value here
    "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter", // Set the format of data that is written to kafka. As the connector parses source cluster messages as binary arrays, this will be the usual required value here
    "connector.class": "com.comcast.kafka.connect.kafka.KafkaSourceConnector", // The full class name of this connector
    "source.bootstrap.servers": "kafka.bootstrap.server1:9092,kafka.bootstrap.server2:9093", // Kafka boostrap servers for source cluster
    "source.topic.whitelist": "test.topic.*", // Mirror topics matching this regex
    "source.auto.offset.reset": "earliest", // For partitions without existing offsets stored, start at the head of the partition
    "source.group.id": "kafka-connect-testing", // Use this group ID when commiting offsets to the source cluster
    "connector.consumer.reconnect.backoff.max.ms": "10000" // Override the default consumer setting "reconnect.backoff.max.ms"
  }
}