| # Licensed to the Apache Software Foundation (ASF) under one 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. |
| |
| from ducktape.services.service import Service |
| from ducktape.utils.util import wait_until |
| |
| from kafkatest.services.kafka.directory import kafka_dir |
| from kafkatest.services.security.security_config import SecurityConfig |
| |
| import os |
| import subprocess |
| |
| """ |
| 0.8.2.1 MirrorMaker options |
| |
| Option Description |
| ------ ----------- |
| --abort.on.send.failure <Stop the Configure the mirror maker to exit on |
| entire mirror maker when a send a failed send. (default: true) |
| failure occurs> |
| --blacklist <Java regex (String)> Blacklist of topics to mirror. |
| --consumer.config <config file> Embedded consumer config for consuming |
| from the source cluster. |
| --consumer.rebalance.listener <A The consumer rebalance listener to use |
| custom rebalance listener of type for mirror maker consumer. |
| ConsumerRebalanceListener> |
| --help Print this message. |
| --message.handler <A custom message Message handler which will process |
| handler of type every record in-between consumer and |
| MirrorMakerMessageHandler> producer. |
| --message.handler.args <Arguments Arguments used by custom rebalance |
| passed to message handler listener for mirror maker consumer |
| constructor.> |
| --num.streams <Integer: Number of Number of consumption streams. |
| threads> (default: 1) |
| --offset.commit.interval.ms <Integer: Offset commit interval in ms (default: |
| offset commit interval in 60000) |
| millisecond> |
| --producer.config <config file> Embedded producer config. |
| --rebalance.listener.args <Arguments Arguments used by custom rebalance |
| passed to custom rebalance listener listener for mirror maker consumer |
| constructor as a string.> |
| --whitelist <Java regex (String)> Whitelist of topics to mirror. |
| """ |
| |
| |
| class MirrorMaker(Service): |
| |
| # Root directory for persistent output |
| PERSISTENT_ROOT = "/mnt/mirror_maker" |
| LOG_DIR = os.path.join(PERSISTENT_ROOT, "logs") |
| LOG_FILE = os.path.join(LOG_DIR, "mirror_maker.log") |
| LOG4J_CONFIG = os.path.join(PERSISTENT_ROOT, "tools-log4j.properties") |
| PRODUCER_CONFIG = os.path.join(PERSISTENT_ROOT, "producer.properties") |
| CONSUMER_CONFIG = os.path.join(PERSISTENT_ROOT, "consumer.properties") |
| |
| logs = { |
| "mirror_maker_log": { |
| "path": LOG_FILE, |
| "collect_default": True} |
| } |
| |
| def __init__(self, context, num_nodes, source, target, whitelist=None, blacklist=None, num_streams=1, |
| new_consumer=False, consumer_timeout_ms=None, offsets_storage="kafka", |
| offset_commit_interval_ms=60000): |
| """ |
| MirrorMaker mirrors messages from one or more source clusters to a single destination cluster. |
| |
| Args: |
| context: standard context |
| source: source Kafka cluster |
| target: target Kafka cluster to which data will be mirrored |
| whitelist: whitelist regex for topics to mirror |
| blacklist: blacklist regex for topics not to mirror |
| num_streams: number of consumer threads to create; can be a single int, or a list with |
| one value per node, allowing num_streams to be the same for each node, |
| or configured independently per-node |
| consumer_timeout_ms: consumer stops if t > consumer_timeout_ms elapses between consecutive messages |
| offsets_storage: used for consumer offsets.storage property |
| offset_commit_interval_ms: how frequently the mirror maker consumer commits offsets |
| """ |
| super(MirrorMaker, self).__init__(context, num_nodes=num_nodes) |
| self.log_level = "DEBUG" |
| self.new_consumer = new_consumer |
| self.consumer_timeout_ms = consumer_timeout_ms |
| self.num_streams = num_streams |
| if not isinstance(num_streams, int): |
| # if not an integer, num_streams should be configured per-node |
| assert len(num_streams) == num_nodes |
| self.whitelist = whitelist |
| self.blacklist = blacklist |
| self.source = source |
| self.target = target |
| |
| self.offsets_storage = offsets_storage.lower() |
| if not (self.offsets_storage in ["kafka", "zookeeper"]): |
| raise Exception("offsets_storage should be 'kafka' or 'zookeeper'. Instead found %s" % self.offsets_storage) |
| |
| self.offset_commit_interval_ms = offset_commit_interval_ms |
| |
| def start_cmd(self, node): |
| cmd = "export LOG_DIR=%s;" % MirrorMaker.LOG_DIR |
| cmd += " export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\";" % MirrorMaker.LOG4J_CONFIG |
| cmd += " export KAFKA_OPTS=%s;" % self.security_config.kafka_opts |
| cmd += " /opt/%s/bin/kafka-run-class.sh kafka.tools.MirrorMaker" % kafka_dir(node) |
| cmd += " --consumer.config %s" % MirrorMaker.CONSUMER_CONFIG |
| cmd += " --producer.config %s" % MirrorMaker.PRODUCER_CONFIG |
| cmd += " --offset.commit.interval.ms %s" % str(self.offset_commit_interval_ms) |
| if isinstance(self.num_streams, int): |
| cmd += " --num.streams %d" % self.num_streams |
| else: |
| # config num_streams separately on each node |
| cmd += " --num.streams %d" % self.num_streams[self.idx(node) - 1] |
| if self.whitelist is not None: |
| cmd += " --whitelist=\"%s\"" % self.whitelist |
| if self.blacklist is not None: |
| cmd += " --blacklist=\"%s\"" % self.blacklist |
| if self.new_consumer: |
| cmd += " --new.consumer" |
| |
| cmd += " 1>> %s 2>> %s &" % (MirrorMaker.LOG_FILE, MirrorMaker.LOG_FILE) |
| return cmd |
| |
| def pids(self, node): |
| try: |
| cmd = "ps ax | grep -i MirrorMaker | grep java | grep -v grep | awk '{print $1}'" |
| pid_arr = [pid for pid in node.account.ssh_capture(cmd, allow_fail=True, callback=int)] |
| return pid_arr |
| except (subprocess.CalledProcessError, ValueError) as e: |
| return [] |
| |
| def alive(self, node): |
| return len(self.pids(node)) > 0 |
| |
| def start_node(self, node): |
| node.account.ssh("mkdir -p %s" % MirrorMaker.PERSISTENT_ROOT, allow_fail=False) |
| node.account.ssh("mkdir -p %s" % MirrorMaker.LOG_DIR, allow_fail=False) |
| |
| self.security_config = self.source.security_config.client_config() |
| self.security_config.setup_node(node) |
| |
| # Create, upload one consumer config file for source cluster |
| consumer_props = self.render("mirror_maker_consumer.properties") |
| consumer_props += str(self.security_config) |
| |
| node.account.create_file(MirrorMaker.CONSUMER_CONFIG, consumer_props) |
| self.logger.info("Mirrormaker consumer props:\n" + consumer_props) |
| |
| # Create, upload producer properties file for target cluster |
| producer_props = self.render('mirror_maker_producer.properties') |
| producer_props += str(self.security_config) |
| self.logger.info("Mirrormaker producer props:\n" + producer_props) |
| node.account.create_file(MirrorMaker.PRODUCER_CONFIG, producer_props) |
| |
| |
| # Create and upload log properties |
| log_config = self.render('tools_log4j.properties', log_file=MirrorMaker.LOG_FILE) |
| node.account.create_file(MirrorMaker.LOG4J_CONFIG, log_config) |
| |
| # Run mirror maker |
| cmd = self.start_cmd(node) |
| self.logger.debug("Mirror maker command: %s", cmd) |
| node.account.ssh(cmd, allow_fail=False) |
| wait_until(lambda: self.alive(node), timeout_sec=10, backoff_sec=.5, |
| err_msg="Mirror maker took to long to start.") |
| self.logger.debug("Mirror maker is alive") |
| |
| def stop_node(self, node, clean_shutdown=True): |
| node.account.kill_process("java", allow_fail=True, clean_shutdown=clean_shutdown) |
| wait_until(lambda: not self.alive(node), timeout_sec=10, backoff_sec=.5, |
| err_msg="Mirror maker took to long to stop.") |
| |
| def clean_node(self, node): |
| if self.alive(node): |
| self.logger.warn("%s %s was still alive at cleanup time. Killing forcefully..." % |
| (self.__class__.__name__, node.account)) |
| node.account.kill_process("java", clean_shutdown=False, allow_fail=True) |
| node.account.ssh("rm -rf %s" % MirrorMaker.PERSISTENT_ROOT, allow_fail=False) |
| self.security_config.clean_node(node) |
