[Bug]: Processing time timers are fired after the window closed in Dataflow

[baeminbo]

What happened?

The pipeline has a DoFn to set a processing time timer in 30 seconds for each input element and add the input value to a bag state. The source generates an element every second, and the window size is 60s with 20s lateness allowed.

In a test with Dataflow Java Legacy Runner, I expected the processing timer wouldn't be fired since the window was closed after the late element arrived. But, actually, the processing timer was fired and the state was empty.

I believe the timers should be cleaned up, the same as states are cleaned up.

FYI. This issue also happens in Runner v2. b/366283709

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: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner