[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)

[github-actions]

This issue has been marked as stale due to 150 days of inactivity. It will be closed in 30 days if no further activity occurs. If you think that’s incorrect or this issue still needs to be addressed, please simply write any comment. If closed, you can reopen the issue at any time. Thank you for your contributions.

[kennknowles]

This should still be handled somehow, if it hasn't been addressed.