[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 && https://github.com/apache/beam/pull/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 https://github.com/apache/beam/pull/14419

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

Issue Priority

Priority: 3 (minor)

Issue Components

• [ ] Component: Python SDK
• [X] 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)