RunRequests with the same run_key emitted in the same sensor evaluation are not de-duplicated

[Gw1p]

What's the issue?

If a sensor yields multiple RunRequests within a single sensor evaluation, Dagster will launch multiple jobs with the same run_key.

What did you expect to happen?

Based on the response I got here, this doesn't seem to be intentional behavior.

I can see 2 approaches that Dagster can take where either would probably be better than the current state:

• sensor should de-duplicate the run_keys emitted within a single sensor evaluation. If the run configs differ, it would just take the first one that the sensor yields and uses that to launch a job
• if sensor detects that there are duplicated run_keys, it raises an error and records Failure in the sensor tick

How to reproduce?

Min example (python 3.11.1; dagster 1.9.5):

import time
from dagster import job, op, Output, sensor, RunRequest

@op
def dummy_op(context):
    context.log.info("Hello")
    yield Output(None)


@job
def dummy_job():
    dummy_op()


@sensor(
    job=dummy_job
)
def dummy_sensor(context):
    now = time.time()
    for i in range(3):
        yield RunRequest(
            run_key=f"run_key_{now}",
        )

On every sensor evalution, dummy_sensor will launch 3 jobs (each with the same run key).

Dagster version

1.9.5

Deployment type

Local

Deployment details

I've experienced this in a dagster helm chart deployment as well as a local one.

Additional information

No response

Message from the maintainers

Impacted by this issue? Give it a 👍! We factor engagement into prioritization.