Add UnboundedCountingSource::to for bounded reads from an UnboundedCountingSource

[arunpandianp]

UnboundedCountingSource currently does not support configuring an end limit, this PR adds that capability.

GenerateSequence with a from + to + rate boils down to a BoundedReadFromUnboundedSource. BoundedReadFromUnboundedSource splits emit all outputs from a single bundle that does not work well on Streaming jobs.

UnboundedCountingSource::to is currently not exposed outside CountingSource. Eventually GenerateSequence can use UnboundedCountingSource::to instead of BoundedReadFromUnboundedSource

[gemini-code-assist]

Summary of Changes

Hello, 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 enhances the UnboundedCountingSource by adding the ability to specify an exclusive end limit for sequence generation. This change allows users to perform bounded reads from an unbounded source, which is a necessary step toward simplifying how GenerateSequence operates in streaming jobs. The implementation includes necessary updates to the source configuration, reader advancement logic, and watermark handling to ensure consistent behavior.

Highlights

• Added end limit support: Introduced an 'end' parameter to UnboundedCountingSource to allow for bounded reads, enabling more flexible sequence generation.
• Updated reader logic: Modified UnboundedCountingReader to respect the new end limit and properly signal completion by setting the watermark to TIMESTAMP_MAX_VALUE.
• Expanded test coverage: Added new test cases in CountingSourceTest to verify correct behavior for bounded sequences with and without rate limiting.

New Features

🧠 You can now enable Memory (public preview) 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.

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 the 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 counterproductive. 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.

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. ↩

[gemini-code-assist]

Code Review

This pull request introduces an exclusive limit (end) to UnboundedCountingSource in CountingSource, allowing unbounded counting sources to stop at a specified value. It adds a .to(long end) builder method, updates serialization/equality methods, and modifies the reader to stop advancing and output a maximum watermark once the limit is reached. The review feedback correctly identifies a critical issue with the introduction of the mutable done state in UnboundedCountingReader. Since done is not checkpointed, restoring the reader from a checkpoint would cause watermark regressions. The reviewer suggests removing the done field entirely and instead deterministically calculating whether the reader is finished using current, source.stride, and source.end.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Instead of maintaining a mutable done state which is not checkpointed (and would cause watermark regressions upon reader restore), we can determine if the reader is done deterministically based on current and source.end. This allows us to completely remove the done field.

Suggested change

	private long current;
	private boolean done = false;
	// Initialized on first advance()
	private long current;
	// Initialized on first advance()

[gemini-code-assist]

Since we are removing the done field, we don't need to set done = true here. We can just return false directly when the next value reaches or exceeds the end limit.

      if (nextValue >= source.end) {
        return false;
      }

[gemini-code-assist]

We can deterministically check if the reader is done by comparing current with source.end - source.stride. If current >= source.end - source.stride, then any subsequent value would be >= source.end, meaning no more elements can be produced. This avoids watermark regression when restoring from a checkpoint.

      return (current >= source.end - source.stride)
          ? BoundedWindow.TIMESTAMP_MAX_VALUE
          : source.timestampFn.apply(current);