Add retry logic for transient commit failures in KafkaCommitOffset

[junaiddshaukat]

Resolves #37606

Summary

KafkaCommitOffset.CommitOffsetDoFn.commitSync() currently catches all
exceptions and silently skips them, meaning transient failures (network
timeouts, temporary broker issues) cause offsets to never be committed.
This can lead to duplicate processing after a restart.

This PR addresses the existing // TODO: consider retrying. by adding
exponential backoff retry (3 retries: 500ms, 1s, 2s) for Kafka's
RetriableException subclasses, while preserving warn-and-skip for
non-retriable exceptions. Retry count is consistent with Beam's existing
Kafka producer default (RETRIES_CONFIG = 3).

Changes

KafkaCommitOffset.java

• Retry up to 3 times with exponential backoff (500ms, 1s, 2s) when
  commitSync() throws a RetriableException (e.g., TimeoutException)
• Non-retriable exceptions still log a warning and skip (preserving
  existing behavior)
• Thread interruption during backoff is handled gracefully

KafkaCommitOffsetTest.java

• Added testCommitOffsetRetriableErrorSucceedsAfterRetry: verifies
  that transient failures are retried and the offset is committed on
  the successful attempt
• Added KafkaRetriableMockConsumer: mock that throws
  RetriableCommitFailedException N times before succeeding

Test plan

• Existing KafkaCommitOffsetTest tests pass
• New retry test verifies commit succeeds after transient failures
• Existing error test confirms non-retriable exceptions still
  warn-and-skip

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[gemini-code-assist]

Summary of Changes

Hello @junaiddshaukat, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request significantly enhances the robustness of Kafka offset committing within Apache Beam by introducing a retry mechanism with exponential backoff. This change addresses the issue where transient failures could lead to silently skipped offset commits and subsequent duplicate data processing. By retrying operations that encounter retriable exceptions, the system becomes more resilient to temporary network or broker issues, ensuring greater data consistency without altering the handling of permanent errors.

Highlights

• Retry Logic: Implemented an exponential backoff retry mechanism for transient Kafka commit failures in KafkaCommitOffset.CommitOffsetDoFn.commitSync().
• Error Handling: Differentiated between retriable and non-retriable exceptions during offset commits, applying retries only for RetriableException subclasses while preserving existing warn-and-skip behavior for others.
• Test Coverage: Added a new test case (testCommitOffsetRetriableErrorSucceedsAfterRetry) and a specialized mock consumer (KafkaRetriableMockConsumer) to thoroughly validate the new retry functionality.

🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Changelog

• sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaCommitOffset.java
  • Imported org.apache.kafka.common.errors.RetriableException for specific error handling.
  • Defined MAX_RETRIES (3) and INITIAL_BACKOFF_MS (500) constants to configure the retry mechanism.
  • Wrapped the consumer.commitSync() call within a for loop to implement up to 3 retries.
  • Introduced exponential backoff (500ms, 1s, 2s) using Thread.sleep() between retries specifically for RetriableException types.
  • Added detailed logging for retry attempts and final failure messages.
  • Gracefully handled InterruptedException during the backoff period.
  • Maintained the existing warn-and-skip behavior for non-retriable Exception types.
• sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaCommitOffsetTest.java
  • Imported java.util.concurrent.atomic.AtomicInteger and org.apache.kafka.clients.consumer.RetriableCommitFailedException for new test utilities.
  • Instantiated KafkaRetriableMockConsumer to simulate retriable commit failures in tests.
  • Added a new test method, testCommitOffsetRetriableErrorSucceedsAfterRetry, to verify that CommitOffsetDoFn successfully commits after encountering transient retriable errors.
  • Implemented KafkaRetriableMockConsumer as a private static inner class, which is designed to throw RetriableCommitFailedException a specified number of times before allowing a successful commit.

Activity

• The pull request resolves issue [Feature Request]: Add retry logic for transient commit failures in KafkaCommitOffset #37606.
• Existing KafkaCommitOffsetTest tests were confirmed to pass.
• A new retry test was added and verified to ensure commits succeed after transient failures.
• Existing error tests were confirmed to show that non-retriable exceptions still warn and skip.
• The author completed the checklist item for mentioning the appropriate issue in the description.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[github-actions]

Assigning reviewers:

R: @ahmedabu98 for label java.
R: @johnjcasey for label kafka.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).