[Bug]: DoFn invoker cache hits repeatedly resolve generic type descriptors

[lofifnc]

What happened?

Beam 2.72 added input and output type descriptors to ByteBuddyDoFnInvokerFactory's generated-constructor cache key in #37355. That fixed a real correctness problem for different generic instantiations of the same DoFn class.

However, newByteBuddyInvoker resolves both descriptors before every constructor-cache lookup:

inputType = fn.getInputTypeDescriptor();
outputType = fn.getOutputTypeDescriptor();

Some runners create a new invoker at every bundle boundary while reusing the same pooled DoFn instance. Beam's default descriptor methods perform reflective generic-type resolution, so every cache hit still repeats that work.

We observed this in a production-scale Java streaming pipeline after upgrading from Beam 2.71 to 2.72:

• TypeDescriptor construction accounted for 30.9% of sampled CPU.
• ByteBuddyDoFnInvokerFactory.newByteBuddyInvoker accounted for 32.6%.
• Bundle-start work accounted for 35.8%.
• With identical application source, resources, input, output, and throughput, a coherent Beam 2.71 comparison reduced worker CPU from 91-92% to 59-60%.
• Total sampled CPU fell from 67.03 s to 41.35 s.
• The Beam 2.71 job processed 22.7% more bundles per minute, so the improvement was not caused by larger or fewer bundles.

A small reproduction is to override the descriptor methods with counters and call DoFnInvokers.invokerFor(fn) twice for the same instance. On current master, each counter is incremented twice even though the generated constructor is already cached.

Expected behavior:

• Preserve the (DoFn class, input type, output type) constructor-cache key from Fix DoFnInvoker cache collision for generic types #37355.
• Select the constructor once for a live DoFn instance and reuse it for later invoker creation.
• Resolve descriptors again for a new/deserialized DoFn instance, including after worker replacement or a job update.
• Do not retain unused DoFn instances.

I have a patch using a weak, identity-keyed per-instance constructor cache. The value is a Constructor<?> and does not reference the DoFn instance. The existing generic-type-aware constructor cache remains unchanged. Tests cover both repeated calls for one instance and separate equal instances with different generic descriptors.

Related: #37351 and #37355.

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: Prism Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[Abacn]

Some runners create a new invoker at every bundle boundary while reusing the same pooled DoFn instance.

Could you please elaborate which runner is involved? While #39310 sounds a valid fix, It might also worth check this direction such that avoid creating a new invoker at every bundle