'Start a cluster as a subprocess.'

def start(self, extra_configs=None):

    self.tmpdir = tempfile.mkdtemp()
    if (not extra_configs):
        extra_configs = {}
    def tmppath(filename):
        'Creates    paths    in    tmpdir.'
        return os.path.join(self.tmpdir, filename)
    LOGGER.info(('Using    temporary    directory:    %s' % self.tmpdir))
    in_conf_dir = tmppath('in-conf')
    os.mkdir(in_conf_dir)
    self.log_dir = tmppath('logs')
    os.mkdir(self.log_dir)
    f = file(os.path.join(in_conf_dir, 'hadoop-metrics.properties'), 'w')
    try:
        f.write('\ndfs.class=org.apache.hadoop.metrics.spi.NoEmitMetricsContext\nmapred.class=org.apache.hadoop.metrics.spi.NoEmitMetricsContext\njvm.class=org.apache.hadoop.metrics.spi.NoEmitMetricsContext\nrpc.class=org.apache.hadoop.metrics.spi.NoEmitMetricsContext\n')
    finally:
        f.close()
    if (self.superuser not in TEST_USER_GROUP_MAPPING):
        TEST_USER_GROUP_MAPPING[self.superuser] = [self.superuser]
    _write_static_group_mapping(TEST_USER_GROUP_MAPPING, tmppath('ugm.properties'))
    core_configs = {('hadoop.proxyuser.%s.groups' % (self.superuser,)): 'users,supergroup', ('hadoop.proxyuser.%s.hosts' % (self.superuser,)): 'localhost', 'mapred.jobtracker.plugins': CLUSTER_JT_PLUGINS}
    extra_configs.update(STARTUP_CONFIGS)
    write_config(core_configs, tmppath('in-conf/core-site.xml'))
    write_config({'mapred.jobtracker.taskScheduler': CLUSTER_TASK_SCHEDULER, 'mapred.queue.names': CLUSTER_QUEUE_NAMES}, tmppath('in-conf/mapred-site.xml'))
    hadoop_policy_keys = ['client', 'client.datanode', 'datanode', 'inter.datanode', 'namenode', 'inter.tracker', 'job.submission', 'task.umbilical', 'refresh.policy', 'admin.operations']
    hadoop_policy_config = {}
    for policy in hadoop_policy_keys:
        hadoop_policy_config[(('security.' + policy) + '.protocol.acl')] = '*'
    write_config(hadoop_policy_config, tmppath('in-conf/hadoop-policy.xml'))
    details_file = file(tmppath('details.json'), 'w+')
    try:
        args = [os.path.join(hadoop.conf.HADOOP_MR1_HOME.get(), 'bin', 'hadoop'), 'jar', hadoop.conf.HADOOP_TEST_JAR.get(), 'minicluster', '-writeConfig', tmppath('config.xml'), '-writeDetails', tmppath('details.json'), '-datanodes', str(self.num_datanodes), '-tasktrackers', str(self.num_tasktrackers), '-useloopbackhosts', '-D', ('hadoop.tmp.dir=%s' % self.tmpdir), '-D', ('mapred.local.dir=%s/mapred/local' % self.tmpdir), '-D', 'mapred.system.dir=/mapred/system', '-D', 'mapred.temp.dir=/mapred/temp', '-D', 'jobclient.completion.poll.interval=100', '-D', 'jobclient.progress.monitor.poll.interval=100', '-D', 'fs.checkpoint.period=1', '-D', ('dfs.secondary.http.address=0.0.0.0:%d' % python_util.find_unused_port()), '-D', ('dfs.thrift.address=localhost:%d' % python_util.find_unused_port()), '-D', ('jobtracker.thrift.address=localhost:%d' % python_util.find_unused_port()), '-D', 'jobclient.completion.poll.interval=50', '-D', 'hadoop.security.authorization=true', '-D', ('hadoop.policy.file=%s/hadoop-policy.xml' % in_conf_dir)]
        for (key, value) in extra_configs.iteritems():
            args.append('-D')
            args.append(((key + '=') + value))
        env = {}
        env['HADOOP_CONF_DIR'] = in_conf_dir
        env['HADOOP_OPTS'] = ('-Dtest.build.data=%s' % (self.tmpdir,))
        env['HADOOP_CLASSPATH'] = ':'.join([hadoop.conf.HADOOP_TEST_JAR.get(), hadoop.conf.HADOOP_PLUGIN_CLASSPATH.get(), os.path.join(os.path.dirname(__file__), 'test_jars', '*')])
        env['HADOOP_HEAPSIZE'] = '128'
        env['HADOOP_HOME'] = hadoop.conf.HADOOP_MR1_HOME.get()
        env['HADOOP_LOG_DIR'] = self.log_dir
        env['USER'] = self.superuser
        if ('JAVA_HOME' in os.environ):
            env['JAVA_HOME'] = os.environ['JAVA_HOME']
        if DEBUG_HADOOP:
            env['HADOOP_OPTS'] = (env.get('HADOOP_OPTS', '') + '    -Xdebug    -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=9999')
        if USE_STDERR:
            stderr = sys.stderr
        else:
            stderr = file(tmppath('stderr'), 'w')
        LOGGER.debug(('Starting    minicluster:    %s    env:    %s' % (repr(args), repr(env))))
        self.clusterproc = subprocess.Popen(args=args, stdout=file(tmppath('stdout'), 'w'), stderr=stderr, env=env)
        details = {}
        start = time.time()
        while (not details):
            try:
                details_file.seek(0)
                details = json.load(details_file)
            except ValueError:
                pass
            if ((self.clusterproc.poll() is not None) or ((not DEBUG_HADOOP) and ((time.time() - start) > MAX_CLUSTER_STARTUP_TIME))):
                LOGGER.debug(('stdout:' + file(tmppath('stdout')).read()))
                if (not USE_STDERR):
                    LOGGER.debug(('stderr:' + file(tmppath('stderr')).read()))
                self.stop()
                raise Exception('Cluster    process    quit    or    is    taking    too    long    to    start.        Aborting.')
    finally:
        details_file.close()
    LOGGER.debug('Successfully    started    minicluster')
    for (k, v) in details.iteritems():
        setattr(self, k, v)
    config = lxml.etree.parse(tmppath('config.xml'))
    self.config = dict(((property.find('./name').text, property.find('./value').text) for property in config.xpath('/configuration/property')))
    self.config_dir = tmppath('conf')
    os.mkdir(self.config_dir)
    hadoop.conf.HADOOP_CONF_DIR.set_for_testing(self.config_dir)
    write_config(self.config, tmppath('conf/core-site.xml'), ['fs.defaultFS', 'jobclient.completion.poll.interval', 'dfs.namenode.checkpoint.period', 'dfs.namenode.checkpoint.dir', (('hadoop.proxyuser.' + self.superuser) + '.groups'), (('hadoop.proxyuser.' + self.superuser) + '.hosts')])
    write_config(self.config, tmppath('conf/hdfs-site.xml'), ['fs.defaultFS', 'dfs.namenode.http-address', 'dfs.namenode.secondary.http-address'])
    write_config({'mapred.job.tracker': ('localhost:%d' % self.jobtracker_port)}, tmppath('conf/mapred-site.xml'))
    write_config(hadoop_policy_config, tmppath('conf/hadoop-policy.xml'))
    args = [hadoop.conf.HADOOP_BIN.get(), '--config', self.config_dir, 'secondarynamenode']
    LOGGER.debug(('Starting    2NN    at:    ' + self.config['dfs.secondary.http.address']))
    LOGGER.debug(('2NN    command:    %s    env:    %s' % (repr(args), repr(env))))
    self.secondary_proc = subprocess.Popen(args=args, stdout=file(tmppath('stdout.2nn'), 'w'), stderr=file(tmppath('stderr.2nn'), 'w'), env=env)
    while True:
        try:
            response = urllib2.urlopen(urllib2.Request(('http://' + self.config['dfs.secondary.http.address'])))
        except urllib2.URLError:
            if ((self.secondary_proc.poll() is not None) or ((not DEBUG_HADOOP) and ((time.time() - start) > MAX_CLUSTER_STARTUP_TIME))):
                LOGGER.debug(('stdout:' + file(tmppath('stdout')).read()))
                if (not USE_STDERR):
                    LOGGER.debug(('stderr:' + file(tmppath('stderr')).read()))
                self.stop()
                raise Exception('2nn    process    quit    or    is    taking    too    long    to    start.    Aborting.')
                break
            else:
                time.sleep(1)
                continue
        response.close()
        break
    LOGGER.debug('Successfully    started    2NN')
