Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

```Traceback (most recent call last):
  File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/engine/flow_runner.py", line 569, in get_flow_run_state
    executors.prepare_upstream_states_for_mapping(
  File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/utilities/executors.py", line 682, in prepare_upstream_states_for_mapping
    value = upstream_state.result[i]
KeyError: 0
└── 17:51:25 | ERROR   | Unexpected error occured in FlowRunner: KeyError(0)```
any help what this error means?

```with prefect.Flow(
    name=flow_config.scheduler_flow_name,
    run_config=run_configs.UniversalRun(
        labels=flow_config.labels,
    ),
    terminal_state_handler=workflow_utils.workflow_terminal_state_handler,
    schedule=schedule,
) as scheduler_flow:

    workflow_id = register_workflow.register_workflow(
        flow_config.scheduler_flow_name,
    )

    # fetch all consumer configs for this provider/resource_type
    consumer_configs = extract_utils.fetch_consumer_configs_for_provider_resource(
        flow_config.provider_code,
        flow_config.resource_type,
    )

    # for each config, start a child flow that runs the ETL process
    # <https://docs-v1.prefect.io/core/idioms/flow-to-flow.html>
    mapped_flow_ids = create_flow_run.map(
        parameters=dict(
            consumer_code=consumer_configs,
            provider_code=prefect.unmapped(flow_config.provider_code),
            resource_type=prefect.unmapped(flow_config.resource_type),
            file_type=prefect.unmapped(flow_config.file_type),
            flow_labels=prefect.unmapped(flow_config.labels),
        ),
        flow_name=prefect.unmapped(
            EtlActionEnum.PIPELINE.value,
        ),
        project_name=prefect.unmapped(
            flow_config.global_config.project_name,
        ),
        labels=prefect.unmapped(flow_config.labels),
    )

    # wait for the mapped Flows to complete before exiting
    wait_for_flow_run.map(
        mapped_flow_ids,
        raise_final_state=prefect.unmapped(True),
    )```

when i comment out the parameters in the create_flow_run the error goes away, but it's not clear to me what is wrong with the parameters

looking at some past threads, seems that parameters wants an iterable. so do i need to make that dict im trying to pass an iterable in a task?

^ that approach worked but surely that couldn't be the best solution

Hey Jon, yeah parameters when mapping is expecting an iterable this is another one of those use cases we solve for in 2.0 since again it can require a bit of finagling to get it to work.