Function start_migration_tool

system_test/utils/kafka_system_test_utils.py:1690–1752

(systemTestEnv, testcaseEnv, onlyThisEntityId=None)

def start_migration_tool(systemTestEnv, testcaseEnv, onlyThisEntityId=None):
    clusterConfigList = systemTestEnv.clusterEntityConfigDictList
    migrationToolConfigList = system_test_utils.get_dict_from_list_of_dicts(clusterConfigList, "role", "migration_tool")

    for migrationToolConfig in migrationToolConfigList:

        entityId = migrationToolConfig["entity_id"]

        if onlyThisEntityId is None or entityId == onlyThisEntityId:

            host              = migrationToolConfig["hostname"]
            jmxPort           = migrationToolConfig["jmx_port"]
            role              = migrationToolConfig["role"]
            kafkaHome         = system_test_utils.get_data_by_lookup_keyval(clusterConfigList, "entity_id", entityId, "kafka_home")
            javaHome          = system_test_utils.get_data_by_lookup_keyval(clusterConfigList, "entity_id", entityId, "java_home")
            jmxPort           = system_test_utils.get_data_by_lookup_keyval(clusterConfigList, "entity_id", entityId, "jmx_port")
            kafkaRunClassBin  = kafkaHome + "/bin/kafka-run-class.sh"

            logger.info("starting kafka migration tool", extra=d)
            migrationToolLogPath     = get_testcase_config_log_dir_pathname(testcaseEnv, "migration_tool", entityId, "default")
            migrationToolLogPathName = migrationToolLogPath + "/migration_tool.log"
            testcaseEnv.userDefinedEnvVarDict["migrationToolLogPathName"] = migrationToolLogPathName

            testcaseConfigsList = testcaseEnv.testcaseConfigsList
            numProducers    = system_test_utils.get_data_by_lookup_keyval(testcaseConfigsList, "entity_id", entityId, "num.producers")
            numStreams      = system_test_utils.get_data_by_lookup_keyval(testcaseConfigsList, "entity_id", entityId, "num.streams")
            producerConfig  = system_test_utils.get_data_by_lookup_keyval(testcaseConfigsList, "entity_id", entityId, "producer.config")
            consumerConfig  = system_test_utils.get_data_by_lookup_keyval(testcaseConfigsList, "entity_id", entityId, "consumer.config")
            zkClientJar     = system_test_utils.get_data_by_lookup_keyval(testcaseConfigsList, "entity_id", entityId, "zkclient.01.jar")
            kafka07Jar      = system_test_utils.get_data_by_lookup_keyval(testcaseConfigsList, "entity_id", entityId, "kafka.07.jar")
            whiteList       = system_test_utils.get_data_by_lookup_keyval(testcaseConfigsList, "entity_id", entityId, "whitelist")
            logFile         = system_test_utils.get_data_by_lookup_keyval(testcaseConfigsList, "entity_id", entityId, "log_filename")

            cmdList = ["ssh " + host,
                       "'JAVA_HOME=" + javaHome,
                       "JMX_PORT=" + jmxPort,
                       kafkaRunClassBin + " kafka.tools.KafkaMigrationTool",
                       "--whitelist="        + whiteList,
                       "--num.producers="    + numProducers,
                       "--num.streams="      + numStreams,
                       "--producer.config="  + systemTestEnv.SYSTEM_TEST_BASE_DIR + "/" + producerConfig,
                       "--consumer.config="  + systemTestEnv.SYSTEM_TEST_BASE_DIR + "/" + consumerConfig,
                       "--zkclient.01.jar="  + systemTestEnv.SYSTEM_TEST_BASE_DIR + "/" + zkClientJar,
                       "--kafka.07.jar="     + systemTestEnv.SYSTEM_TEST_BASE_DIR + "/" + kafka07Jar,
                       " &> " + migrationToolLogPath + "/migrationTool.log",
                       " & echo pid:$! > " + migrationToolLogPath + "/entity_" + entityId + "_pid'"]

            cmdStr = " ".join(cmdList)
            logger.debug("executing command: [" + cmdStr + "]", extra=d)
            system_test_utils.async_sys_call(cmdStr)
            time.sleep(5)

            pidCmdStr = "ssh " + host + " 'cat " + migrationToolLogPath + "/entity_" + entityId + "_pid' 2> /dev/null"
            logger.debug("executing command: [" + pidCmdStr + "]", extra=d)
            subproc = system_test_utils.sys_call_return_subproc(pidCmdStr)

            # keep track of the remote entity pid in a dictionary
            for line in subproc.stdout.readlines():