[
https://issues.apache.org/jira/browse/BEAM-9132?focusedWorklogId=377737&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-377737
]
ASF GitHub Bot logged work on BEAM-9132:
----------------------------------------
Author: ASF GitHub Bot
Created on: 27/Jan/20 17:01
Start Date: 27/Jan/20 17:01
Worklog Time Spent: 10m
Work Description: mxm commented on pull request #10694: [BEAM-9132] Use a
unique bundle id across all SDK workers bound to the same environment
URL: https://github.com/apache/beam/pull/10694
We have observed these errors in a state-intense application:
```
Error processing instruction 107. Original traceback is
Traceback (most recent call last):
File "apache_beam/runners/common.py", line 780, in
apache_beam.runners.common.DoFnRunner.process
File "apache_beam/runners/common.py", line 587, in
apache_beam.runners.common.PerWindowInvoker.invoke_process
File "apache_beam/runners/common.py", line 659, in
apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window
File "apache_beam/runners/common.py", line 880, in
apache_beam.runners.common._OutputProcessor.process_outputs
File "apache_beam/runners/common.py", line 895, in
apache_beam.runners.common._OutputProcessor.process_outputs
File "redacted.py", line 56, in process
recent_events_map = load_recent_events_map(recent_events_state)
File "redacted.py", line 128, in _load_recent_events_map
items_in_recent_events_bag = list(recent_events_state.read())
File "apache_beam/runners/worker/bundle_processor.py", line 335, in
__iter__
for elem in self.first:
File "apache_beam/runners/worker/bundle_processor.py", line 214, in
__iter__
self._state_key, self._coder_impl, is_cached=self._is_cached)
File "apache_beam/runners/worker/sdk_worker.py", line 692, in blocking_get
self._materialize_iter(state_key, coder))
File "apache_beam/runners/worker/sdk_worker.py", line 723, in
_materialize_iter
self._underlying.get_raw(state_key, continuation_token)
File "apache_beam/runners/worker/sdk_worker.py", line 603, in get_raw
continuation_token=continuation_token)))
File "apache_beam/runners/worker/sdk_worker.py", line 637, in
_blocking_request
raise RuntimeError(response.error)
RuntimeError: Unknown process bundle instruction id '107'
```
Notice that the error is thrown on the Runner side in the GrpcStateService.
Explanation: The error occurs when multiple ExecutableStages are scheduled on
the same TaskManager. Depending on how many SDK workers are available, they
may
end up sharing the same environment. This means that they will also share the
same GrpcStateService. The state service uses the bundle id to reference the
StateRequestHandler of the currently active bundle. The problem is that the
code
used a from zero ascending id generator for each stage, even if it shared the
same environment.
The bundle id generator always start from 0 which eventually leads to a race
condition between the state requests of the transforms which share the same
environments. Their ids will overlap leading to falsely removing state
handlers
of other stages in GrpcStateService.
Solution: Use a unique id generator per environment which is shared by all
the
stages within an environment.
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/)
| --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/)
| --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/)
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)
| --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/)
| --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/)
XLang | --- | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/)
| --- | --- | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://builds.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/)
| --- | ---
See
[.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md)
for trigger phrase, status and link of all Jenkins jobs.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 377737)
Time Spent: 1h (was: 50m)
> State request handler is removed prematurely when closing ActiveBundle
> ----------------------------------------------------------------------
>
> Key: BEAM-9132
> URL: https://issues.apache.org/jira/browse/BEAM-9132
> Project: Beam
> Issue Type: Bug
> Components: java-fn-execution
> Reporter: Maximilian Michels
> Assignee: Maximilian Michels
> Priority: Major
> Time Spent: 1h
> Remaining Estimate: 0h
>
> We have observed these errors in a state-intense application:
> {noformat}
> Error processing instruction 107. Original traceback is
> Traceback (most recent call last):
> File "apache_beam/runners/common.py", line 780, in
> apache_beam.runners.common.DoFnRunner.process
> File "apache_beam/runners/common.py", line 587, in
> apache_beam.runners.common.PerWindowInvoker.invoke_process
> File "apache_beam/runners/common.py", line 659, in
> apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window
> File "apache_beam/runners/common.py", line 880, in
> apache_beam.runners.common._OutputProcessor.process_outputs
> File "apache_beam/runners/common.py", line 895, in
> apache_beam.runners.common._OutputProcessor.process_outputs
> File "redacted.py", line 56, in process
> recent_events_map = load_recent_events_map(recent_events_state)
> File "redacted.py", line 128, in _load_recent_events_map
> items_in_recent_events_bag = list(recent_events_state.read())
> File "apache_beam/runners/worker/bundle_processor.py", line 335, in __iter__
> for elem in self.first:
> File "apache_beam/runners/worker/bundle_processor.py", line 214, in __iter__
> self._state_key, self._coder_impl, is_cached=self._is_cached)
> File "apache_beam/runners/worker/sdk_worker.py", line 692, in blocking_get
> self._materialize_iter(state_key, coder))
> File "apache_beam/runners/worker/sdk_worker.py", line 723, in
> _materialize_iter
> self._underlying.get_raw(state_key, continuation_token)
> File "apache_beam/runners/worker/sdk_worker.py", line 603, in get_raw
> continuation_token=continuation_token)))
> File "apache_beam/runners/worker/sdk_worker.py", line 637, in
> _blocking_request
> raise RuntimeError(response.error)
> RuntimeError: Unknown process bundle instruction id '107'
> {noformat}
> Notice that the error is thrown on the Runner side. It seems to originate
> from the {{ActiveBundle}} de-registering the state request handler too early
> when the processing may still be going on in the SDK Harness.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
