heron/executor/src/python/heron_executor.py
def _get_tmaster_processes(self):
    ''' get the command to start the tmaster processes '''
    retval = {}
    tmaster_cmd_lst = [
        self.tmaster_binary,
        '--topology_name=%s' % self.topology_name,
        '--topology_id=%s' % self.topology_id,
        '--zkhostportlist=%s' % self.state_manager_connection,
        '--zkroot=%s' % self.state_manager_root,
        '--myhost=%s' % self.master_host,
        '--master_port=%s' % str(self.master_port),
        '--controller_port=%s' % str(self.tmaster_controller_port),
        '--stats_port=%s' % str(self.tmaster_stats_port),
        '--config_file=%s' % self.heron_internals_config_file,
        '--override_config_file=%s' % self.override_config_file,
        '--metrics_sinks_yaml=%s' % self.metrics_sinks_config_file,
        '--metricsmgr_port=%s' % str(self.metrics_manager_port),
        '--ckptmgr_port=%s' % str(self.checkpoint_manager_port)]

    tmaster_env = self.shell_env.copy() if self.shell_env is not None else {}
    tmaster_cmd = Command(tmaster_cmd_lst, tmaster_env)
    if os.environ.get('ENABLE_HEAPCHECK') is not None:
      tmaster_cmd.env.update({
          'LD_PRELOAD': "/usr/lib/libtcmalloc.so",
          'HEAPCHECK': "normal"
      })

    retval["heron-tmaster"] = tmaster_cmd

    if self.metricscache_manager_mode.lower() != "disabled":
      retval["heron-metricscache"] = self._get_metrics_cache_cmd()

    if self.health_manager_mode.lower() != "disabled":
      retval["heron-healthmgr"] = self._get_healthmgr_cmd()

    retval[self.metricsmgr_ids[0]] = self._get_metricsmgr_cmd(
        self.metricsmgr_ids[0],
        self.metrics_sinks_config_file,
        self.metrics_manager_port)

    if self.is_stateful_topology:
      retval.update(self._get_ckptmgr_process())

    return retval
def _get_tmaster_processes(self):
    ''' get the command to start the tmaster processes '''
    retval = {}
    tmaster_cmd_lst = [
        self.tmaster_binary,
        '--topology_name=%s' % self.topology_name,
        '--topology_id=%s' % self.topology_id,
        '--zkhostportlist=%s' % self.state_manager_connection,
        '--zkroot=%s' % self.state_manager_root,
        '--myhost=%s' % self.master_host,
        '--master_port=%s' % str(self.master_port),
        '--controller_port=%s' % str(self.tmaster_controller_port),
        '--stats_port=%s' % str(self.tmaster_stats_port),
        '--config_file=%s' % self.heron_internals_config_file,
        '--override_config_file=%s' % self.override_config_file,
        '--metrics_sinks_yaml=%s' % self.metrics_sinks_config_file,
        '--metricsmgr_port=%s' % str(self.metrics_manager_port),
        '--ckptmgr_port=%s' % str(self.checkpoint_manager_port)]

    tmaster_env = self.shell_env.copy() if self.shell_env is not None else {}
    tmaster_cmd = Command(tmaster_cmd_lst, tmaster_env)
    if os.environ.get('ENABLE_HEAPCHECK') is not None:
      tmaster_cmd.env.update({
          'LD_PRELOAD': "/usr/lib/libtcmalloc.so",
          'HEAPCHECK': "normal"
      })

    retval["heron-tmaster"] = tmaster_cmd

    if self.metricscache_manager_mode.lower() != "disabled":
      retval["heron-metricscache"] = self._get_metrics_cache_cmd()

    if self.health_manager_mode.lower() != "disabled":
      retval["heron-healthmgr"] = self._get_healthmgr_cmd()

    retval[self.metricsmgr_ids[0]] = self._get_metricsmgr_cmd(
        self.metricsmgr_ids[0],
        self.metrics_sinks_config_file,
        self.metrics_manager_port)

    if self.is_stateful_topology:
      retval.update(self._get_ckptmgr_process())

    return retval
get the command to start the tmaster processes