[Feature Request]: Remove dependency of KafkaIO withDynamicRead on beam_fn_api

[damnMeddlingKid]

What happened?

The withDynamicRead method was gated with the beam_fn_api experimental flag in https://issues.apache.org/jira/browse/BEAM-11946 && #14168.

From reading the ticket it sounds like the intention was to ensure that we read from Kafka with an SDF when using x-lang features.

Im wondering if this is still necessary today because reading from Kafka via SDF is the default behaviour and there are a few exceptional cases where it falls back to an unbounded source .

heres a link to a previous PR that removed beam_fn_api from the rest of KafkaIO but kept it for dynamic reads #14419

cc: @kennknowles you might have some context on if this is still required ?

Issue Priority

Priority: 3 (minor)

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

[johnjcasey]

I believe we still need this, because we cannot fall back to the legacy implementation if the user needs dynamic reads

[kennknowles]

The way this should work is that the transform should expand for dynamic reads and the runner can say yes/no to whether it can execute it. We should definitely eliminate the beam_fn_api experiment entirely. It is really old and was meant for running stuff in a nonstandard way before it was really ready for production. TBH we probably never needed it IMO (because there are better ways to achieve the goal)