[Feature Request]: Add retry logic for transient commit failures in KafkaCommitOffset

[junaiddshaukat]

What would you like to happen?

What needs to happen?

In KafkaCommitOffset.CommitOffsetDoFn.processElement(), when
consumer.commitSync() fails, the exception is caught and logged
as a warning but no retry is attempted:

https://github.com/apache/beam/blob/master/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaCommitOffset.java#L87-L90

There is an existing TODO comment: // TODO: consider retrying.

Transient failures (e.g., TimeoutException,
RetriableCommitFailedException) during offset commits can cause
offsets to not be committed, potentially leading to duplicate
processing after a restart. Adding retry with backoff for retriable
exceptions would improve reliability.

Proposed approach

• Add retry with exponential backoff for retriable Kafka exceptions
• Keep current warn-and-skip behavior for non-retriable exceptions
• Use a bounded number of retries (e.g., 3 attempts) to avoid
  blocking indefinitely

Issue Priority

Priority: 3 (nice-to-have improvement)

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

[junaiddshaukat]

.take-issue