':param local_task_job: The local task job associated with running the
associated task instance.
:type local_task_job: airflow.jobs.LocalTaskJob'

def __init__(self, local_task_job):

    self._task_instance = local_task_job.task_instance
    popen_prepend = []
    cfg_path = None
    if self._task_instance.run_as_user:
        self.run_as_user = self._task_instance.run_as_user
    else:
        try:
            self.run_as_user = conf.get('core', 'default_impersonation')
        except conf.AirflowConfigException:
            self.run_as_user = None
    if (self.run_as_user and (self.run_as_user != getpass.getuser())):
        self.logger.debug('Planning    to    run    as    the    {}    user'.format(self.run_as_user))
        cfg_dict = conf.as_dict(display_sensitive=True)
        cfg_subset = {'core': cfg_dict.get('core', {}), 'smtp': cfg_dict.get('smtp', {}), 'scheduler': cfg_dict.get('scheduler', {}), 'webserver': cfg_dict.get('webserver', {})}
        (temp_fd, cfg_path) = mkstemp()
        subprocess.call(['sudo', 'chown', self.run_as_user, cfg_path])
        subprocess.call(['sudo', 'chmod', '600', cfg_path])
        with os.fdopen(temp_fd, 'w') as temp_file:
            json.dump(cfg_subset, temp_file)
        popen_prepend = ['sudo', '-H', '-u', self.run_as_user]
    self._cfg_path = cfg_path
    self._command = (popen_prepend + self._task_instance.command_as_list(raw=True, ignore_all_deps=local_task_job.ignore_all_deps, ignore_depends_on_past=local_task_job.ignore_depends_on_past, ignore_ti_state=local_task_job.ignore_ti_state, pickle_id=local_task_job.pickle_id, mark_success=local_task_job.mark_success, job_id=local_task_job.id, pool=local_task_job.pool, cfg_path=cfg_path))
    self.process = None
