[Bug]: KafkaIO metrics write per-partition values to per-split metrics

[sjvanrossum]

What happened?

#31137 overwrites the per-split metric backlog_bytes.${SPLIT} with a per-partition value rather than the accumulated value for the split. #31281 introduces a Map to store metrics for all past and current splits (1 partition) of the ReadFromKafkaDoFn instance and may repeatedly overwrite non-current splits with stale values. The map used to store these values is not thread-safe and may trigger a ConcurrentModificationException since GetSize and other SDF methods may concurrently attempt to read and write the map. Finally, the per-split caches kept by the instance are keyed on TopicPartition, which is not unique among all splits since the split may override the bootstrap server.

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

[sjvanrossum]

Fix provided in #32921.

[damccorm]

@sjvanrossum it looks like automation added this to the 2.61.0 release, is it actually a release blocker?

[Abacn]

close after #32921 merged