def submit_job(manager, job_config):
    """ Launch new job from specified config. May have been previously 'setup'
    if 'setup_params' in job_config is empty.
    """
    # job_config is raw dictionary from JSON (from MQ or HTTP endpoint).
    job_id = job_config.get('job_id')
    try:
        command_line = job_config.get('command_line')

        setup_params = job_config.get('setup_params', {})
        force_setup = job_config.get('setup')
        remote_staging = job_config.get('remote_staging', {})
        dependencies_description = job_config.get('dependencies_description', None)
        env = job_config.get('env', [])
        submit_params = job_config.get('submit_params', {})
        touch_outputs = job_config.get('touch_outputs', [])
        job_config = None
        if setup_params or force_setup:
            input_job_id = setup_params.get("job_id", job_id)
            tool_id = setup_params.get("tool_id", None)
            tool_version = setup_params.get("tool_version", None)
            use_metadata = setup_params.get("use_metadata", False)
            job_config = setup_job(
                manager,
                input_job_id,
                tool_id,
                tool_version,
                use_metadata,
            )

        if job_config is not None:
            job_directory = job_config["job_directory"]
            jobs_directory = os.path.abspath(os.path.join(job_directory, os.pardir))
            command_line = command_line.replace('__PULSAR_JOBS_DIRECTORY__', jobs_directory)

        # TODO: Handle __PULSAR_JOB_DIRECTORY__ config files, metadata files, etc...
        manager.touch_outputs(job_id, touch_outputs)
        launch_config = {
            "remote_staging": remote_staging,
            "command_line": command_line,
            "dependencies_description": dependencies_description,
            "submit_params": submit_params,
            "env": env,
            "setup_params": setup_params,
        }
        manager.preprocess_and_launch(job_id, launch_config)
    except Exception:
        manager.handle_failure_before_launch(job_id)
        raise
Launch new job from specified config. May have been previously 'setup'
    if 'setup_params' in job_config is empty.