You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "shunping (via GitHub)" <gi...@apache.org> on 2026/06/30 18:32:31 UTC
[PR] Fix lost state updates in Set state [beam]
shunping opened a new pull request, #39175:
URL: https://github.com/apache/beam/pull/39175
When elements are added to `SynchronousSetRuntimeState`, it occasionally triggers data compaction (`_compact_data`) which launches asynchronous `clear` and `extend` requests to the State Handler. However, these futures were not tracked. The `commit()` method only blocked on the last futures generated during the commit phase, causing outstanding compaction futures to complete out of order or after the bundle processor finished, resulting in lost state updates.
An example failed test:
https://github.com/apache/beam/actions/runs/28001498086/job/82874623596
Traceback:
```
_______ StatefulDoFnOnDirectRunnerTest.test_stateful_set_state_portably ________
[gw2] linux -- Python 3.10.20 /runner/_work/beam/beam/sdks/python/test-suites/tox/py310/build/srcs/sdks/python/target/.tox-py310/py310/bin/python
self = <apache_beam.transforms.userstate_test.StatefulDoFnOnDirectRunnerTest testMethod=test_stateful_set_state_portably>
def test_stateful_set_state_portably(self):
class SetStatefulDoFn(beam.DoFn):
SET_STATE = SetStateSpec('buffer', VarIntCoder())
def process(self, element, set_state=beam.DoFn.StateParam(SET_STATE)):
_, value = element
aggregated_value = 0
set_state.add(value)
for saved_value in set_state.read():
aggregated_value += saved_value
yield aggregated_value
> with TestPipeline() as p:
apache_beam/transforms/userstate_test.py:715:
...
if duration:
state_thread = threading.Thread(
target=functools.partial(self._observe_state, message_thread),
name='wait_until_finish_state_observer')
state_thread.daemon = True
state_thread.start()
start_time = time.time()
duration_secs = duration / 1000
while (time.time() - start_time < duration_secs and
state_thread.is_alive()):
time.sleep(1)
else:
self._observe_state(message_thread)
if self._runtime_exception:
> raise self._runtime_exception
E RuntimeError: Pipeline job-045 failed in state FAILED: bundle inst010 stage-009 failed:Traceback (most recent call last):
E File "apache_beam/runners/common.py", line 1499, in apache_beam.runners.common.DoFnRunner.process
E return self.do_fn_invoker.invoke_process(windowed_value)
E File "apache_beam/runners/common.py", line 913, in apache_beam.runners.common.PerWindowInvoker.invoke_process
E self._invoke_process_per_window(
E File "apache_beam/runners/common.py", line 1058, in apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window
E self.process_method(*args_for_process, **kwargs_for_process),
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py310/build/srcs/sdks/python/target/.tox-py310/py310/lib/python3.10/site-packages/apache_beam/transforms/core.py", line 2126, in <lambda>
E wrapper = lambda x, *args, **kwargs: [fn(x, *args, **kwargs)]
E File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py310/build/srcs/sdks/python/target/.tox-py310/py310/lib/python3.10/site-packages/apache_beam/testing/util.py", line 202, in _equal
E raise BeamAssertException(msg)
E apache_beam.testing.util.BeamAssertException: Failed assert: [1, 3, 6, 10, 10] == [1, 4, 6, 3, 7], unexpected elements [4, 7], missing elements [10, 10]
E
```
This PR fixes a race condition in `SynchronousSetRuntimeState` where asynchronous state requests triggered by state compaction or early clears were not being awaited.
- Introduced `self._futures` to record all asynchronous state requests initiated by compaction and early clears.
- Updated `commit()` to combine `self._futures` with current commit-time futures and await all of them before completing.
- Added a regression test.
--
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] Fix lost state updates in Set state [beam]
Posted by "jrmccluskey (via GitHub)" <gi...@apache.org>.
jrmccluskey merged PR #39175:
URL: https://github.com/apache/beam/pull/39175
--
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] Fix lost state updates in Set state [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #39175:
URL: https://github.com/apache/beam/pull/39175#discussion_r3502215205
##########
sdks/python/apache_beam/runners/worker/bundle_processor.py:
##########
@@ -678,15 +682,26 @@ def clear(self) -> None:
self._added_elements = set()
def commit(self) -> None:
- to_await = None
if self._cleared:
- to_await = self._state_handler.clear(self._state_key)
+ self._futures.append(self._state_handler.clear(self._state_key))
+ self._cleared = False
if self._added_elements:
- to_await = self._state_handler.extend(
- self._state_key, self._value_coder.get_impl(), self._added_elements)
- if to_await:
- # To commit, we need to wait on the last state request future to complete.
- to_await.get()
+ self._futures.append(
+ self._state_handler.extend(
+ self._state_key,
+ self._value_coder.get_impl(),
+ self._added_elements))
+ self._added_elements = set()
+
+ # Block on all outstanding async state requests to ensure data is committed.
+ # We must swap and clear self._futures before awaiting them. Awaiting a future
+ # yields control, during which new futures could be appended to self._futures.
+ all_futures = self._futures
+ self._futures = []
+
+ for f in all_futures:
+ if f:
+ f.get()
Review Comment:
The current implementation swaps `self._futures` into `all_futures` and clears `self._futures` before awaiting them. However, as the comment correctly points out, awaiting a future yields control, during which new futures could be appended to `self._futures` (e.g., via concurrent state updates or callbacks). Since the loop only iterates over the snapshot `all_futures`, any newly appended futures in `self._futures` will be left unawaited when `commit()` returns. This can lead to lost state updates or out-of-order execution.
To ensure all outstanding futures (including those transitively added during the await phase) are fully resolved before `commit()` returns, we should use a `while` loop to drain `self._futures` until it is completely empty.
```suggestion
# Block on all outstanding async state requests to ensure data is committed.
# We must swap and clear self._futures before awaiting them. Awaiting a future
# yields control, during which new futures could be appended to self._futures.
while self._futures:
all_futures = self._futures
self._futures = []
for f in all_futures:
if f:
f.get()
```
--
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] Fix lost state updates in Set state [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #39175:
URL: https://github.com/apache/beam/pull/39175#issuecomment-4848468340
Assigning reviewers:
R: @jrmccluskey for label python.
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).
--
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] Fix lost state updates in Set state [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #39175:
URL: https://github.com/apache/beam/pull/39175#issuecomment-4848415136
## 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 addresses a race condition in `SynchronousSetRuntimeState` where asynchronous state requests were not being properly tracked or awaited. By ensuring all outstanding state operations are completed during the commit phase, the change prevents data loss caused by compaction futures finishing out of order or after the bundle processor has already finalized.
### Highlights
* **Asynchronous State Tracking**: Introduced a `_futures` list in `SynchronousSetRuntimeState` to track all asynchronous state requests triggered by compaction and early clears.
* **Commit Synchronization**: Updated the `commit()` method to aggregate and await all outstanding asynchronous futures, ensuring that state updates are fully processed before the bundle completes.
* **Regression Testing**: Added a new test case `test_stateful_set_state_compaction_race_portably` that simulates latency in state requests to verify the fix for the identified race condition.
<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] Fix lost state updates in Set state [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #39175:
URL: https://github.com/apache/beam/pull/39175#discussion_r3502225174
##########
sdks/python/apache_beam/runners/worker/bundle_processor.py:
##########
@@ -678,15 +682,26 @@ def clear(self) -> None:
self._added_elements = set()
def commit(self) -> None:
- to_await = None
if self._cleared:
- to_await = self._state_handler.clear(self._state_key)
+ self._futures.append(self._state_handler.clear(self._state_key))
+ self._cleared = False
if self._added_elements:
- to_await = self._state_handler.extend(
- self._state_key, self._value_coder.get_impl(), self._added_elements)
- if to_await:
- # To commit, we need to wait on the last state request future to complete.
- to_await.get()
+ self._futures.append(
+ self._state_handler.extend(
+ self._state_key,
+ self._value_coder.get_impl(),
+ self._added_elements))
+ self._added_elements = set()
+
+ # Block on all outstanding async state requests to ensure data is committed.
+ # We must swap and clear self._futures before awaiting them. Awaiting a future
+ # yields control, during which new futures could be appended to self._futures.
+ all_futures = self._futures
+ self._futures = []
+
+ for f in all_futures:
+ if f:
+ f.get()
Review Comment:
Nope. Those futures will not be lost and will be handled in the next commit.
--
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