You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "scwhittle (via GitHub)" <gi...@apache.org> on 2026/06/24 09:47:04 UTC
[PR] [Dataflow Java Streaming] Add a timeout to how long commits will retry to the service. [beam]
scwhittle opened a new pull request, #39085:
URL: https://github.com/apache/beam/pull/39085
In cases where the service is unavailable this prevents build-up of commits on workers that are no longer relevant. This defaults to 30 minutes but can be disabled via an experiment or by setting the option.
------------------------
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](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Java Streaming] Add a timeout to how long commits will retry to the service. [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #39085:
URL: https://github.com/apache/beam/pull/39085#discussion_r3466903562
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -224,14 +228,49 @@ public void onResponse(StreamingCommitResponse response) {
failureHandler.throwIfNonEmpty();
}
- @Override
@SuppressWarnings("ReferenceEquality")
+ private boolean belongsToThisHandler(StreamAndRequest streamAndRequest) {
+ return streamAndRequest.handler == this;
+ }
+
+ @Override
public boolean hasPendingRequests() {
- return pending.entrySet().stream().anyMatch(e -> e.getValue().handler == this);
+ return pending.entrySet().stream().anyMatch(e -> belongsToThisHandler(e.getValue()));
}
@Override
+ @SuppressWarnings("ReferenceEquality")
public void onDone(Status status) {
+ if (maxRetryDuration.compareTo(Duration.ZERO) > 0) {
+ // Remove the requests that have exceeded the retry time so they are not retried.
+ long nowNanos = System.nanoTime();
+ long maxRetryDurationNanos = maxRetryDuration.toNanos();
+ Iterator<Map.Entry<Long, StreamAndRequest>> iterator = pending.entrySet().iterator();
+ int keptRequests = 0, removedRequests = 0;
+ while (iterator.hasNext()) {
+ StreamAndRequest streamAndRequest = checkNotNull(iterator.next().getValue());
+ PendingRequest pendingRequest = streamAndRequest.request;
+ if (!belongsToThisHandler(streamAndRequest)
+ || nowNanos - pendingRequest.getStartTimeNanos() < maxRetryDurationNanos) {
+ ++keptRequests;
+ continue;
+ }
+ ++removedRequests;
+ iterator.remove();
+ try {
+ pendingRequest.completeWithStatus(CommitStatus.ABORTED);
+ } catch (RuntimeException e) {
+ // Ignore exceptions and retry the commit.
+ LOG.warn("Exception while aborting commit due to retry timeout.", e);
+ }
Review Comment:
The comment `// Ignore exceptions and retry the commit.` is misleading. Since `iterator.remove()` was already called on line 259, the commit has been removed from the `pending` map and will **not** be retried. The comment should be updated to clarify that we are ignoring the exception and continuing to abort other expired commits.
```suggestion
} catch (RuntimeException e) {
// Ignore exceptions and continue aborting other expired commits.
LOG.warn("Exception while aborting commit due to retry timeout.", e);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Java Streaming] Add a timeout to how long commits will retry to the service. [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #39085:
URL: https://github.com/apache/beam/pull/39085#discussion_r3466139826
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowStreamingPipelineOptions.java:
##########
@@ -196,6 +196,13 @@ public interface DataflowStreamingPipelineOptions extends PipelineOptions {
void setStuckCommitDurationMillis(int value);
+ @Description(
+ "Retry commits on stream errors until this much time has elapsed since the commit was scheduled. If zero, retry forever.")
+ @Default.InstanceFactory(CommitWorkStreamRetryTimeoutMillisFactory.class)
+ long getCommitWorkStreamRetryTimeoutMillis();
+
+ void getCommitWorkStreamRetryTimeoutMillis(long value);
Review Comment:
The setter method is incorrectly named `getCommitWorkStreamRetryTimeoutMillis` instead of `setCommitWorkStreamRetryTimeoutMillis`. This violates JavaBean naming conventions and will cause issues with `PipelineOptions` serialization and command-line parsing.
```suggestion
void setCommitWorkStreamRetryTimeoutMillis(long value);
```
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowStreamingPipelineOptions.java:
##########
@@ -343,4 +350,15 @@ public Boolean create(PipelineOptions options) {
return ExperimentalOptions.hasExperiment(options, "enable_windmill_service_direct_path");
}
}
+
+ /** defaults to false unless one of the experiment is set. */
Review Comment:
The Javadoc comment states that the option `defaults to false`, but the factory actually returns a `Long` representing 30 minutes in milliseconds (or `0L` if disabled). Update the comment to accurately describe the default value.
```suggestion
/** Defaults to 30 minutes unless the disable_commit_retry_timeout experiment is set. */
```
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -224,14 +228,50 @@ public void onResponse(StreamingCommitResponse response) {
failureHandler.throwIfNonEmpty();
}
- @Override
@SuppressWarnings("ReferenceEquality")
+ private boolean belongsToThisHandler(StreamAndRequest streamAndRequest) {
+ return streamAndRequest.handler == this;
+ }
+
+ @Override
public boolean hasPendingRequests() {
- return pending.entrySet().stream().anyMatch(e -> e.getValue().handler == this);
+ return pending.entrySet().stream().anyMatch(e -> belongsToThisHandler(e.getValue()));
}
@Override
+ @SuppressWarnings("ReferenceEquality")
public void onDone(Status status) {
+ if (maxRetryDuration.compareTo(Duration.ZERO) > 0) {
+ // Remove the requests that have exceeded the retry time so they are not retried.
+ long startTimeRetryThresholdMsec = System.currentTimeMillis() - maxRetryDuration.toMillis();
+ Iterator<Map.Entry<Long, StreamAndRequest>> iterator = pending.entrySet().iterator();
+ int keptRequests = 0, removedRequests = 0;
+ while (iterator.hasNext()) {
+ StreamAndRequest streamAndRequest = checkNotNull(iterator.next().getValue());
+ PendingRequest pendingRequest = streamAndRequest.request;
+ if (!belongsToThisHandler(streamAndRequest)
+ || pendingRequest.getStartTimeMillis() > startTimeRetryThresholdMsec) {
+ ++keptRequests;
+ continue;
Review Comment:
Using `System.currentTimeMillis()` to measure elapsed time or timeouts is susceptible to system clock adjustments (e.g., NTP syncs). It is highly recommended to use a monotonic clock like `System.nanoTime()` for calculating elapsed time.
```suggestion
long maxRetryNanos = maxRetryDuration.toNanos();
long nowNanos = System.nanoTime();
Iterator<Map.Entry<Long, StreamAndRequest>> iterator = pending.entrySet().iterator();
int keptRequests = 0, removedRequests = 0;
while (iterator.hasNext()) {
StreamAndRequest streamAndRequest = checkNotNull(iterator.next().getValue());
PendingRequest pendingRequest = streamAndRequest.request;
if (!belongsToThisHandler(streamAndRequest)
|| (nowNanos - pendingRequest.getStartTimeNanos()) <= maxRetryNanos) {
++keptRequests;
continue;
}
```
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -224,14 +228,50 @@ public void onResponse(StreamingCommitResponse response) {
failureHandler.throwIfNonEmpty();
}
- @Override
@SuppressWarnings("ReferenceEquality")
+ private boolean belongsToThisHandler(StreamAndRequest streamAndRequest) {
+ return streamAndRequest.handler == this;
+ }
+
+ @Override
public boolean hasPendingRequests() {
- return pending.entrySet().stream().anyMatch(e -> e.getValue().handler == this);
+ return pending.entrySet().stream().anyMatch(e -> belongsToThisHandler(e.getValue()));
}
@Override
+ @SuppressWarnings("ReferenceEquality")
public void onDone(Status status) {
+ if (maxRetryDuration.compareTo(Duration.ZERO) > 0) {
+ // Remove the requests that have exceeded the retry time so they are not retried.
+ long startTimeRetryThresholdMsec = System.currentTimeMillis() - maxRetryDuration.toMillis();
+ Iterator<Map.Entry<Long, StreamAndRequest>> iterator = pending.entrySet().iterator();
+ int keptRequests = 0, removedRequests = 0;
+ while (iterator.hasNext()) {
+ StreamAndRequest streamAndRequest = checkNotNull(iterator.next().getValue());
+ PendingRequest pendingRequest = streamAndRequest.request;
+ if (!belongsToThisHandler(streamAndRequest)
+ || pendingRequest.getStartTimeMillis() > startTimeRetryThresholdMsec) {
+ ++keptRequests;
+ continue;
+ }
+ ++removedRequests;
+ try {
+ pendingRequest.completeWithStatus(CommitStatus.ABORTED);
+ } catch (RuntimeException e) {
+ // Catch possible exceptions to ensure that an exception for one commit does not prevent
+ // other commits from being processed. Aggregate all the failures to throw after
+ // processing the response if they exist.
Review Comment:
The comment mentions aggregating all failures to throw after processing, but the catch block only logs a warning and does not aggregate or rethrow any exceptions. Update the comment to reflect the actual behavior.
```suggestion
// Catch possible exceptions to ensure that an exception for one commit does not prevent
// other commits from being processed.
```
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -376,34 +416,46 @@ private void issueMultiChunkRequest(long id, PendingRequest pendingRequest)
}
}
- @AutoValue
- abstract static class PendingRequest {
+ private static class PendingRequest {
+ private final String computationId;
+ private final WorkItemCommitRequest request;
+ private final Consumer<CommitStatus> onDone;
+ private final long startTimeMillis;
- private static PendingRequest create(
+ private PendingRequest(
String computationId, WorkItemCommitRequest request, Consumer<CommitStatus> onDone) {
- return new AutoValue_GrpcCommitWorkStream_PendingRequest(computationId, request, onDone);
+ this.computationId = computationId;
+ this.request = request;
+ this.onDone = onDone;
+ this.startTimeMillis = System.currentTimeMillis();
}
- abstract String computationId();
+ String getComputationId() {
+ return computationId;
+ }
- abstract WorkItemCommitRequest request();
+ WorkItemCommitRequest getRequest() {
+ return request;
+ }
- abstract Consumer<CommitStatus> onDone();
+ long getStartTimeMillis() {
+ return startTimeMillis;
+ }
Review Comment:
To support monotonic time tracking using `System.nanoTime()`, update `PendingRequest` to store and expose `startTimeNanos` instead of `startTimeMillis`.
```java
private final long startTimeNanos;
private PendingRequest(
String computationId, WorkItemCommitRequest request, Consumer<CommitStatus> onDone) {
this.computationId = computationId;
this.request = request;
this.onDone = onDone;
this.startTimeNanos = System.nanoTime();
}
String getComputationId() {
return computationId;
}
WorkItemCommitRequest getRequest() {
return request;
}
long getStartTimeNanos() {
return startTimeNanos;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Java Streaming] Add a timeout to how long commits will retry to the service. [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #39085:
URL: https://github.com/apache/beam/pull/39085#discussion_r3466849490
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -224,14 +228,48 @@ public void onResponse(StreamingCommitResponse response) {
failureHandler.throwIfNonEmpty();
}
- @Override
@SuppressWarnings("ReferenceEquality")
+ private boolean belongsToThisHandler(StreamAndRequest streamAndRequest) {
+ return streamAndRequest.handler == this;
+ }
+
+ @Override
public boolean hasPendingRequests() {
- return pending.entrySet().stream().anyMatch(e -> e.getValue().handler == this);
+ return pending.entrySet().stream().anyMatch(e -> belongsToThisHandler(e.getValue()));
}
@Override
+ @SuppressWarnings("ReferenceEquality")
public void onDone(Status status) {
+ if (maxRetryDuration.compareTo(Duration.ZERO) > 0) {
+ // Remove the requests that have exceeded the retry time so they are not retried.
+ long startTimeRetryThresholdNanos = System.nanoTime() - maxRetryDuration.toNanos();
+ Iterator<Map.Entry<Long, StreamAndRequest>> iterator = pending.entrySet().iterator();
+ int keptRequests = 0, removedRequests = 0;
+ while (iterator.hasNext()) {
+ StreamAndRequest streamAndRequest = checkNotNull(iterator.next().getValue());
+ PendingRequest pendingRequest = streamAndRequest.request;
+ if (!belongsToThisHandler(streamAndRequest)
+ || pendingRequest.getStartTimeNanos() > startTimeRetryThresholdNanos) {
+ ++keptRequests;
+ continue;
+ }
+ ++removedRequests;
+ try {
+ pendingRequest.completeWithStatus(CommitStatus.ABORTED);
+ iterator.remove();
+ } catch (RuntimeException e) {
+ // Ignore exceptions and retry the commit.
+ LOG.warn("Exception while aborting commit due to retry timeout.", e);
+ }
+ }
Review Comment:
There are two issues in this retry timeout cleanup block:
1. **Robustness & Memory Leak**: If `pendingRequest.completeWithStatus(CommitStatus.ABORTED)` throws a `RuntimeException`, `iterator.remove()` is skipped. This leaves the timed-out request in the `pending` map, causing it to be retried on stream reconnect and potentially leaking memory. We should remove the request from the iterator first before invoking the callback.
2. **`System.nanoTime()` Comparison Safety**: Comparing absolute `nanoTime` values directly using `>` can fail if the values wrap around (which can happen on long-running JVMs). Using `System.nanoTime() - pendingRequest.getStartTimeNanos() < maxRetryDurationNanos` is safe against wrap-around and is also much more readable as it directly represents the elapsed time.
```java
if (maxRetryDuration.compareTo(Duration.ZERO) > 0) {
// Remove the requests that have exceeded the retry time so they are not retried.
long maxRetryDurationNanos = maxRetryDuration.toNanos();
Iterator<Map.Entry<Long, StreamAndRequest>> iterator = pending.entrySet().iterator();
int keptRequests = 0, removedRequests = 0;
while (iterator.hasNext()) {
StreamAndRequest streamAndRequest = checkNotNull(iterator.next().getValue());
PendingRequest pendingRequest = streamAndRequest.request;
if (!belongsToThisHandler(streamAndRequest)
|| System.nanoTime() - pendingRequest.getStartTimeNanos() < maxRetryDurationNanos) {
++keptRequests;
continue;
}
++removedRequests;
iterator.remove();
try {
pendingRequest.completeWithStatus(CommitStatus.ABORTED);
} catch (RuntimeException e) {
LOG.warn("Exception while aborting commit due to retry timeout.", e);
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Java Streaming] Add a timeout to how long commits will retry to the service. [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #39085:
URL: https://github.com/apache/beam/pull/39085#issuecomment-4787970745
## Summary of Changes
Hello, I'm Gemini Code Assist[^1]! 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 implements a safety mechanism for Dataflow streaming pipelines by introducing a timeout for commit retries. By limiting how long a commit will be retried when the service is unavailable, the system avoids resource build-up on workers for work that is no longer relevant. The feature is configurable, allowing users to tune the timeout or disable it if necessary.
### Highlights
* **Commit Retry Timeout**: Introduced a configurable timeout for commit retries in Dataflow streaming pipelines to prevent the accumulation of stale commits during service unavailability.
* **Configuration**: Added a new pipeline option 'commitWorkStreamRetryTimeoutMillis' which defaults to 30 minutes and can be adjusted or disabled via experiments.
* **Implementation**: Updated GrpcCommitWorkStream to track commit start times and abort requests that exceed the defined retry duration.
<details>
<summary><b>New Features</b></summary>
🧠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](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
</details>
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
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.
<b>Customization</b>
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](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with :thumbsup: and :thumbsdown: 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](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Java Streaming] Add a timeout to how long commits will retry to the service. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #39085:
URL: https://github.com/apache/beam/pull/39085#issuecomment-4788970513
/gemini review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Java Streaming] Add a timeout to how long commits will retry to the service. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #39085:
URL: https://github.com/apache/beam/pull/39085#issuecomment-4789058788
/gemini review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Dataflow Java Streaming] Add a timeout to how long commits will retry to the service. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #39085:
URL: https://github.com/apache/beam/pull/39085
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org