[Bug]: RunInference with automatic refresh using a global window side inputs takes too long to update sometimes

[AnandInguva]

What happened?

Sometimes, a global window side input takes too long to update on a Dataflow job.

The automatic model refresh feature of RunInference uses a pattern WatchFilePattern which uses a global windowed side input to fetch the latest model path. If the pipeline using this feature launches a dataflow job and there is an update to the model path, it could take long time to update the model since sometimes the workers are busy dealing with backlog.

The code I used to run is at https://github.com/apache/beam/blob/11e2bae4cbee4cc4f9d200a71511d921e8591dcd/examples/notebooks/beam-ml/automatic_model_refresh.ipynb

The work around I found is to increase the num_workers>1, I usually set it to 5.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[kennknowles]

Does this reproduce on runners other than Dataflow? Can we isolate it to the SDK harness (possibly a cache that keeps getting its invalidation re-upped?) or is it runner-specific?

Maybe CC @lostluck regarding Prism's ability to get repro this. (global window triggered side input never updating as long as worker kept busy)

[lostluck]

Triggers are not yet implemented in Prism, so a global window side input will not re-execute. (The pipeline will fail before starting presently.) Triggers are last thing to implement since they have the most complicated effects on downstream computations.

Overall Prism feature status is tracked in #29650

[lostluck]

I don't know how RunInference is implemented, but if it's not adding the trigger itself, then the linked code as written won't execute RunInference at all, since the global window side input will never complete, and the consuming transform won't start until the side input is ready.

[lostluck]

Ok, so it's WatchFilePattern applying a AfterProcessingTime trigger.

[scwhittle]

I believe this is a long-standing bug within the python sdk. Side inputs within the global window are cached in PerWindowInvoker without respecting the side input cache token.

This is part of the bundle processor which is reused across bundles. The side input values are otherwise attempted with reset here or by the runner by modifying the side input cache token.

Since bundle procesors are cached as long as there is a steady rate of input so that the last accessed time is less than 60 seconds here, this can lead to extended periods where the captured global side input value is used without refresh.

I think that we should remove the caching at the invoker level as it does not respect the cache token and the StateBackedSideInput supports caching itself. This may be a performance regression as the state cache is currently disabled by default though.

[kennknowles]

Nice find! I have seen quite a few disconnected issues around side input refresh in Python. I hope this is the real root cause.

[scwhittle]

@AnandInguva Can you verify the above fixes your issue? It does for my separate testing pipeline