[Bug]: Mutable stringset race condition

[Abacn]

What happened?

#32650 introduced a race condition when stringset is getting rate (runner about to report metrics) and adding more values. In particular, on Dataflow legacy runer it may throw runtime Exceptions when reading here:

beam/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricsToCounterUpdateConverter.java

Line 105 in 4f37e53

stringList.setElements(new ArrayList<>(stringSetData.stringSet()));

This prominently happens on Beam 2.60.0 streaming pipeline using FileIO read from an unbounded input of file metadata, and files come in with different directories. In this way stringset add does not happen at once, and likely trigger the race

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

[liferoad]

Let me know if this does not block 2.61.0.