Enable concurrent KedroSession usage

[nejox]

Description

runtime_params OmegaConf resolver registration is process-global, breaking concurrent KedroSession usage

Context

When multiple KedroSession instances run concurrently in the same process (e.g. a FastAPI service triggering pipelines per request), OmegaConfigLoader registers _get_runtime_value as the runtime_params resolver with replace=True on every session creation — meaning Session B silently overwrites the resolver of Session A, causing requests to resolve each other's runtime parameters. In our case we created an API which triggers the kedro pipelines in the same kedro env where we interpolate filepaths based on a runtime_parameter ID which differ by request. Our deployment works with multiple KubernetesPods running one worker and leverage multithreading. While this seems only relevant during KedroCatalog creation (afaik) this became a problem in our project.

Possible Implementation

A backwards-compatible fix (brainstormed with Claude) could be to register the resolver once at module level backed by a contextvars.ContextVar, and set/reset the ContextVar in KedroSession.enter/exit — preserving all existing resolver behaviour (OmegaConf.select, nested params, default values, error handling) while achieving per-session isolation for both threaded and asyncio concurrency models.

[merelcht]

Hi @nejox , thanks for opening this issue. This sounds very relevant to work we've started on allowing multiple runs per session and essentially allowing serving of pipelines. You can find more of our thinking in this milestone: https://github.com/kedro-org/kedro/milestone/87 and I'm also tagging @ankatiyar and @rashidakanchwala who are involved with the work.

@nejox can you maybe share some more details on what you're doing? Or even a small code snippet about how you're running the sessions concurrently?

[nejox]

Hi @merelcht, of course! Our main setup involves kedro in the "classical" way by using pipelines in a batch-fashion and in our case orchestrated via airflow. Additionally, we have some sort of "intermediate updating mechanism" which is achieved by sending an request to an fastapi server which then triggers parts of the updating pipeline like so

@router.get("/update")
def updates(
    customer: str,
    id_store: Optional[List[int]],
    id_product: Optional[List[int]],
):
    from kedro.framework.startup import bootstrap_project
    from kedro.framework.session import KedroSession
  
    extra_params = {"customer": customer}
    with KedroSession.create(env="xy", extra_params=extra_params) as session:
        session.run(pipeline_name="updating", runner=SequentialRunner())

(I admit this might be a hacky way but worked for us quite well)

Our kedro envs are separating our customers into groups and we use interpolation in the catalog based on the runtime parameter customer to build the final filepaths:

03_testdataset:
     type: pandas.ParquetDataset
     filepath: s3://customer-${runtime_param:customer, ${oc.env:CUSTOMER}}/data/03_primary/testdataset.parquet

This works totally fine without multithreading, but recently we changed our fastapi deployment in our kubernetes cluster to multiple pods with only one worker. In this case, then multiple request land on the same worker, multiple (e.g. >2) KedroSessions share the same process and we get a race condition on which request sets the OmegaConf runtime_param resolver the last. This is only relevant during the catalog creation during Kedro with a very small time window but in our case resulted in wrong filepaths built as the runtime_param in process #1 was overwritten by the value from process #2.

I see that our use case is not the usual approach within kedro and this might be not within the scope but having KedroSession support concurrency would be great 😄

[merelcht]

Thank you so much for sharing these details @nejox. We're planning to address this in the work on the KedroSession. We'll work on this iteratively and concurrency is one of the more advanced topics we'll work on a bit later. You can expect some changes to be released in the coming months.