It's unclear how/where I should configure job concurrency in Dagster OSS based on the documentation about dagster

[sam-goodwin]

I am following the docs precisely and getting errors. It's very confusing to understand.

my_job = define_asset_job(
    name="my_jb",
    partitions_def=scan_id,
    executor_def=multiprocess_executor,
    config={
        "execution": {
            "config": {
                "multiprocess": {
                    "tag_concurrency_limits": [
                        {
                            "key": "yarn",
                            "value": "dask",
                            "limit": 3,
                        }
                    ],
                },
            }
        }
    },
)

Error:

Backfill failed for jnpeozgr: dagster._core.errors.DagsterInvalidConfigError: Error in config for job orion_human
    Error 1: Received unexpected config entry "multiprocess" at path root:execution:config. Expected: "{ max_concurrent?: Int? retries?: { disabled?: { } enabled?: { } } start_method?: { forkserver?: { preload_modules?: [String] } spawn?: { } } tag_concurrency_limits?: [{ key: String limit: Int value?: (String | { applyLimitPerUniqueValue: Bool }) }] }".

Docs:

[garethbrickman]

@sam-goodwin Could you try restructuring your code as below?

# Configure the multiprocess_executor with the desired settings
configured_multiprocess_executor = multiprocess_executor.configured(
    {
        "max_concurrent": 3,  # limits concurrent processes to 3
        "tag_concurrency_limits": [
            {
                "key": "yarn",
                "value": "dask",
                "limit": 3,
            }
        ],
    }
)

# Define the job with the configured executor
my_job = define_asset_job(
    name="my_job",
    partitions_def=scan_id,
    executor_def=configured_multiprocess_executor,
)