Hello, I have a flow on prefect cloud that is stuck in a pending state, even though I have an agent ready that was previously working. It says

x slots

used in the late runs section, even though the flow does not have any tags associated with the concurrency limit (I have some other flows that do). Any ideas on how I can resolve this issue?

@Marvin is it better to create deployment with yaml file or with python? Or either is fine?

@Marvin How can I disable prefect logging to stdout

Hi everyone! I want to rerun a prefect flow that failed. I changed a part of my code to debug the fail. If I retry a flow from the UI, would it retry with my updated code? Or would you suggest "copy to new run" to make sure it will run my updated code?

@Marvin I tried to use prefect_test_harness but get error sqlite3.OperationalError: unable to open database file

Hi Community! Has anyone tried to handle an exception from a Future inside another task? I was trying to design a Map Reduce style workflow, where a Mapper spreads tasks and a Reducer collects them all together, and skips failed mappers, but this is the error I get

prefect.exceptions.UnfinishedRun: Run is in PENDING state, its result is not available.

For who is more interested, this the code I am testing

from random import randint

from prefect import task, flow

class DummyError(Exception):
    pass

@task
def say_hello():
    print("Hello, world!")


@task
def t1():
    print("t1")


@task
def t2():
    print("t2")


@task
def t3():
    raise DummyError("t3")


@task
def mapper():
    task_id = randint(1, 2)
    task_instance = globals()[f"t{task_id}"]  # dynamically select a task
    return [
        task_instance.submit(),
        t3.submit()
    ]


@task
def reducer(futures):
    for future in futures:
        try:
            future.result()
        except Exception as e:
            print(f"Error: {e}")


@flow
def myflow():
    say_hello()
    futures = mapper()
    reducer(futures)
    say_hello()
    return "OK"


if __name__ == "__main__":
    result = myflow()

@Marvin got this error message today when launching a flow that was working fine until yesterday

Error creating flow run: Validation failed for field 'env'. Failure reason: '' is not of type 'object'

@Marvin what is this and how best to resolve?

Task parameter introspection took 19.636 seconds , exceeding `PREFECT_TASK_INTROSPECTION_WARN_THRESHOLD` of 10.0. Try wrapping large task parameters with `prefect.utilities.annotations.quote` for increased performance, e.g. `my_task(quote(param))`. To disable this message set `PREFECT_TASK_INTROSPECTION_WARN_THRESHOLD=0`

I'm using DaskTaskRunner, karpenter, and the

dask_kubernetes.KubeCluster

class – trying to get a flow to run on a GPU-enabled node. What's happening: • The pod_template I'm using for the flow specifies

<http://nvidia.com/gpu|nvidia.com/gpu>: 1

, per the docs • karpenter starts a new node, which has a GPU (🙌) • However, the new node can't accommodate the dask client because it doesn't have the right resource annotation ("0/3 nodes are available: … , 3 Insufficient nvidia.com/gpu") I realise this might be a question better-suited to the dask community, but would appreciate any stories of people successfully running Prefect 2 workloads on GPU-accelerated nodes and/or help figuring out the above specific issue.

can one agent run multiple work-queues in prefect 2.0