[Bug]: KafkaIO maintains duplicate caches for record size and offset estimators caches

[sjvanrossum]

What happened?

Offset consumers are created and cached for every KafkaSourceDescriptor processed per ReadFromKafkaDoFn instance. Caching multiple open Kafka consumers for the same KafkaSourceDescriptor is expensive both on the client and server side and may exhaust connection limits on some Kafka clusters.

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 #32928.

[damccorm]

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

[sjvanrossum]

@damccorm it's not a hard blocker, but since #32928 was approved last week (pending merge of #32921, now merged) I figured I'd flag it on the milestone for your awareness in case you'd like to see it included in 2.61.0.

[damccorm]

Closing since #32928 is merged