@josh one more quick question. It looks like if I run a flow via prefect cloud using the k8s agent, a flow job is created and caching via prefect cloud runs properly. However, if I use a KubernetesJobEnvironment which means the job created by the agent creates a secondary job (because I need my flow to run with a custom k8s config), I lose caching. Is it possible this setup leads to in memory caching in the KubernetesJobEnvironment rather than prefect cloud caching?

And then here's logs for the same flow, same task, same inputs, etc. Only change is that I removed the KubernetesJobEnvironment (flow runs in job created from the agent)

logs.txt

Which I think would mean caching basically won't work (the cache will look empty on each run of the flow)

But I’m thinking that the caching just isn’t happening with this turned off maybe

Actually, looking like caching and result handlers are fine. Issue is that no DEBUG level logs are making it to prefect cloud when the flow is run with KubernetesJobEnvironment so I can't see the logs about caching

INFO level logs and above are fine

Hey @josh one last question in this thread: I have a bunch of tasks that write a file locally and name that file with the md5 hash of the file contents. I have a custom result handler I can associate with those tasks which is basically: write(result): upload file at task result local file path to GCS, return GCS URI read(uri): download from GCS URI to local file path, return local file path My issue is when I have 2 tasks back to back that use this result handler and both tasks are setup with caching based on all inputs cache validator. In order to validate inputs for the child task, for each potential cached state, the read method of the result handler for the parent task is called. This leads to downloading from GCS just to validate the returned local file path (which I could return without downloading if able to distinguish between a read call to return a cached output and a read call to validate an input for a child task). My workaround right now is to put a passthrough task between these two with a regular JSONResultHandler that simply returns the argument it receives (

def passthrough(val): return val

), but wondering what pattern you would suggest here (that way rather than downloading a file, I'm only deserializing JSON to check inputs for all cached states)