create_auto_cdc_from_snapshot_flow Python session resolution fails if having multiple snapshot flows

[david_aspegren]

When a pipeline contains more than one create_auto_cdc_from_snapshot_flow flow (each driven by a custom Python next_snapshot_and_version function), flow resolution fails intermittently/consistently with:

RuntimeError: The original Spark session is being accessed instead of the per-flow cloned session during parallel analysis. This is commonly caused by spawning threads inside a flow function that access the Spark session.

I am having functions to figure out the next snapshot like this:
```
def next_x_snapshot_and_version(latest_version):
versions = spark.read.table(SOURCE_TABLE).select("file_modification_time").distinct()
```

Is this a bug?

[ThomasBehne]

This error occurs because your custom `next_x_snapshot_and_version` function references the global `spark` session variable directly, bypassing Delta Live Tables' per-flow cloned session during parallel execution. When running multiple CDC snapshot flows concurrently, DLT isolates each flow using its own cloned session; accessing global `spark` state breaks this thread safety. To fix it, update your function signature to accept a `spark_session` argument explicitly (e.g., `def next_x_snapshot_and_version(spark_session, latest_version) and pass the thread-safe `spark` instance into your custom function via a `lambda` inside your `create_auto_cdc_from_snapshot_flow` call.

[david_aspegren]

Thank you! I am still doing something wrong though, my function is now called like so:

dp.create_auto_cdc_from_snapshot_flow(

    target=SILVER_TABLE,

    source=lambda latest_version: next_locations_snapshot_and_version(spark, latest_version),

    keys=["location_id"],

    stored_as_scd_type=2,

    track_history_except_column_list=["file_modification_time", "source_file", "ingestion_time"],

)

anything more i need to do to get to "pass the thread-safe `spark` instance"