You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "liferoad (via GitHub)" <gi...@apache.org> on 2025/10/15 16:22:58 UTC
[PR] [TEST] increase grpc keepalive timeout and adjust ping settings [beam]
liferoad opened a new pull request, #36528:
URL: https://github.com/apache/beam/pull/36528
Adjust GRPC channel settings to reduce ping frequency and allow more flexible keepalive behavior. This improves performance by reducing unnecessary network traffic while maintaining connection stability.
------------------------
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] optimize grpc settings [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2482017170
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
I think we can wait longer. cc @tvalentyn WDTY?
--
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] optimize grpc settings and fix some flaky tests [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2445089115
##########
sdks/python/apache_beam/io/filebasedsink.py:
##########
@@ -205,6 +205,19 @@ def open_writer(self, init_result, uid):
# We also ensure there will be no collisions with uid and a
# (possibly unsharded) file_path_prefix and a (possibly empty)
# file_name_suffix.
+ from apache_beam.pvalue import EmptySideInput
Review Comment:
https://github.com/apache/beam/issues/36563. Good idea. Let me add this as TODO for this issue.
--
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] [TEST] increase grpc keepalive timeout and adjust ping settings [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2436407574
##########
sdks/python/apache_beam/io/filebasedsink.py:
##########
@@ -205,6 +205,19 @@ def open_writer(self, init_result, uid):
# We also ensure there will be no collisions with uid and a
# (possibly unsharded) file_path_prefix and a (possibly empty)
# file_name_suffix.
+ from apache_beam.pvalue import EmptySideInput
Review Comment:
this should fix:
```
=================================== FAILURES ===================================
_______________________ TestParquet.test_sink_transform ________________________
[gw5] linux -- Python 3.13.3 /runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/bin/python
self = <apache_beam.io.parquetio_test.TestParquet testMethod=test_sink_transform>
def test_sink_transform(self):
with TemporaryDirectory() as tmp_dirname:
path = os.path.join(tmp_dirname + "tmp_filename")
> with TestPipeline() as p:
^^^^^^^^^^^^^^
apache_beam/io/parquetio_test.py:359:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/pipeline.py:670: in __exit__
self.result = self.run()
^^^^^^^^^^
target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/testing/test_pipeline.py:122: in run
state = result.wait_until_finish(duration=self.timeout)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <apache_beam.runners.portability.portable_runner.PipelineResult object at 0x7901b41dc690>
duration = None
def wait_until_finish(self, duration=None):
self.process_method(*args_for_process, **kwargs_for_process),
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/io/iobase.py", line 1431, in process
writer = self.sink.open_writer(init_result, str(uuid.uuid4()))
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/options/value_provider.py", line 193, in _f
return fnc(self, *args, **kwargs)
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/io/filebasedsink.py", line 211, in open_writer
writer_path = FileSystems.join(init_result, uid) + suffix
~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/io/filesystems.py", line 169, in join
filesystem = FileSystems.get_filesystem(basepath)
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/io/filesystems.py", line 155, in get_filesystem
raise BeamIOError('Unable to get the Filesystem', {path: e})
apache_beam.io.filesystem.BeamIOError: Unable to get the Filesystem with exceptions {<apache_beam.pvalue.EmptySideInput object at 0x7901ba06e5d0>: AttributeError("'EmptySideInput' object has no attribute 'strip'")}
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/runners/worker/sdk_worker.py", line 316, in _execute
response = task()
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/runners/worker/sdk_worker.py", line 390, in <lambda>
lambda: self.create_worker().do_instruction(request), request)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/runners/worker/sdk_worker.py", line 662, in do_instruction
return getattr(self, request_type)(
~~~~~~~~~~~~~~~~~~~~~~~~~~~^
getattr(request, request_type), request.instruction_id)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/runners/worker/sdk_worker.py", line 700, in process_bundle
bundle_processor.process_bundle(instruction_id))
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1274, in process_bundle
input_op_by_transform_id[element.transform_id].process_encoded(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
element.data)
^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/runners/worker/bundle_processor.py", line 237, in process_encoded
self.output(decoded_value)
~~~~~~~~~~~^^^^^^^^^^^^^^^
File "apache_beam/runners/worker/operations.py", line 568, in apache_beam.runners.worker.operations.Operation.output
def output(self, windowed_value, output_index=0):
File "apache_beam/runners/worker/operations.py", line 570, in apache_beam.runners.worker.operations.Operation.output
_cast_to_receiver(self.receivers[output_index]).receive(windowed_value)
File "apache_beam/runners/worker/operations.py", line 260, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
def receive(self, windowed_value):
File "apache_beam/runners/worker/operations.py", line 263, in apache_beam.runners.worker.operations.SingletonElementConsumerSet.receive
self.consumer.process(windowed_value)
File "apache_beam/runners/worker/operations.py", line 951, in apache_beam.runners.worker.operations.DoOperation.process
with self.scoped_process_state:
File "apache_beam/runners/worker/operations.py", line 952, in apache_beam.runners.worker.operations.DoOperation.process
delayed_applications = self.dofn_runner.process(o)
File "apache_beam/runners/common.py", line 1500, in apache_beam.runners.common.DoFnRunner.process
self._reraise_augmented(exn, windowed_value)
File "apache_beam/runners/common.py", line 1609, in apache_beam.runners.common.DoFnRunner._reraise_augmented
raise new_exn
File "apache_beam/runners/common.py", line 1498, in apache_beam.runners.common.DoFnRunner.process
return self.do_fn_invoker.invoke_process(windowed_value)
File "apache_beam/runners/common.py", line 912, in apache_beam.runners.common.PerWindowInvoker.invoke_process
self._invoke_process_per_window(
File "apache_beam/runners/common.py", line 1057, in apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window
self.process_method(*args_for_process, **kwargs_for_process),
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/io/iobase.py", line 1431, in process
writer = self.sink.open_writer(init_result, str(uuid.uuid4()))
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/options/value_provider.py", line 193, in _f
return fnc(self, *args, **kwargs)
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/io/filebasedsink.py", line 211, in open_writer
writer_path = FileSystems.join(init_result, uid) + suffix
~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/io/filesystems.py", line 169, in join
filesystem = FileSystems.get_filesystem(basepath)
File "/runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/lib/python3.13/site-packages/apache_beam/io/filesystems.py", line 155, in get_filesystem
raise BeamIOError('Unable to get the Filesystem', {path: e})
apache_beam.io.filesystem.BeamIOError: Unable to get the Filesystem with exceptions {<apache_beam.pvalue.EmptySideInput object at 0x7901ba06e5d0>: AttributeError("'EmptySideInput' object has no attribute 'strip'")} [while running 'WriteToParquet/Write/WriteImpl/WriteBundles'] with exceptions None
```
at least for our tests.
--
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] optimize grpc settings [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2449048203
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
Review Comment:
Would it make sense to use the same constant like `DEFAULT_OPTIONS` to define GRPC client settings in both places?
--
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] optimize grpc settings [beam]
Posted by "sergiitk (via GitHub)" <gi...@apache.org>.
sergiitk commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2511371815
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,20 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 20000ms (20s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
Pro tip: python allows `_` as a separator in numeric literals to make them more readable
```suggestion
("grpc.keepalive_timeout_ms", 600_000),
```
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,14 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
+ # keep 20 seconds for now. This is needed for other options to work.
("grpc.keepalive_time_ms", 20000),
+ # keep 5 minutes for now.
Review Comment:
Nit: phrasing "for now" without any explanation is not very useful. Does it imply it should be changes at some point? If so, when?
--
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] optimize grpc settings [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2482014530
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,12 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 20000ms (20s), increased to 10 minutes for stability
Review Comment:
removed `grpc.keepalive_time_ms` since https://github.com/grpc/grpc/blob/master/doc/keepalive.md#defaults-values
INT_MAX (disabled) on client
--
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] optimize grpc settings [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #36528:
URL: https://github.com/apache/beam/pull/36528#issuecomment-3530834749
thanks @liferoad ! let's see if it helps with the test flakiness.
--
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] optimize grpc settings and fix some flaky tests [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on PR #36528:
URL: https://github.com/apache/beam/pull/36528#issuecomment-3426736413
> i'd suggest to split grpc tune up into a dedicated change from test robustness changes.
Sounds good. Let me split this PR.
--
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] optimize grpc settings [beam]
Posted by "sergiitk (via GitHub)" <gi...@apache.org>.
sergiitk commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2501658416
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,12 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 20000ms (20s), increased to 10 minutes for stability
Review Comment:
It's disabled by default, and unless you enable it, all other keepalive settings you have (`grpc.keepalive_timeout_ms `, `grpc.http2.max_pings_without_data`, `grpc.keepalive_permit_without_calls`) have no effect.
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,12 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 20000ms (20s), increased to 10 minutes for stability
Review Comment:
It's disabled by default, and unless you enable it, all other keepalive settings you have (`grpc.keepalive_timeout_ms`, `grpc.http2.max_pings_without_data`, `grpc.keepalive_permit_without_calls`) have no effect.
--
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] optimize grpc settings [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2449359641
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
Review Comment:
I don't see this documented, I believe it has been deprecated
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
Review Comment:
Are the pings expensive if we only have a single channel? Does this mean that it will take 180s instead of 30s to notice that the runner terminated? This doesn't seem good for streaming pipelines if the case.
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
+ # Default: 2, set to 0 to allow unlimited ping strikes
+ ("grpc.http2.max_ping_strikes", 0),
Review Comment:
think this just affects server, rm?
https://github.com/grpc/grpc/blob/master/doc/keepalive.md
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
+ # Default: 2, set to 0 to allow unlimited ping strikes
+ ("grpc.http2.max_ping_strikes", 0),
+ # Default: 0 (disabled), enable socket reuse for better handling
+ ("grpc.so_reuseport", 1),
+ # Default: 0 (disabled), set 30s TCP timeout for connection control
+ ("grpc.tcp_user_timeout_ms", 30000),
Review Comment:
does python use c-core for grpc? if so it seems that this does default to 20s not 0
where are you seeing the defaults?
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
Review Comment:
think affects server only https://github.com/grpc/grpc/blob/master/doc/keepalive.md
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
+ # Default: 2, set to 0 to allow unlimited ping strikes
+ ("grpc.http2.max_ping_strikes", 0),
+ # Default: 0 (disabled), enable socket reuse for better handling
+ ("grpc.so_reuseport", 1),
Review Comment:
also server only?
--
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] [TEST] optimize grpc settings and fix some flaky tests [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2440010020
##########
sdks/python/apache_beam/transforms/external_test.py:
##########
@@ -247,9 +247,19 @@ def test_pipeline_generation_with_runner_overrides(self):
'in the pipeline')
self.assertEqual(1, len(list(pubsub_read_transform.outputs.values())))
- self.assertEqual(
- list(pubsub_read_transform.outputs.values()),
- list(external_transform.inputs.values()))
+ self.assertEqual(1, len(list(external_transform.inputs.values())))
Review Comment:
```
=================================== FAILURES ===================================
_____ ExternalTransformTest.test_pipeline_generation_with_runner_overrides _____
[gw1] linux -- Python 3.13.3 /runner/_work/beam/beam/sdks/python/test-suites/tox/py313/build/srcs/sdks/python/target/.tox-py313-cloud/py313-cloud/bin/python
self = <apache_beam.transforms.external_test.ExternalTransformTest testMethod=test_pipeline_generation_with_runner_overrides>
@unittest.skipIf(apiclient is None, 'GCP dependencies are not installed')
def test_pipeline_generation_with_runner_overrides(self):
pipeline_properties = [
'--job_name=test-job',
'--project=test-project',
'--temp_location=gs://beam/tmp',
'--no_auth',
'--dry_run=True',
'--sdk_location=container',
'--runner=DataflowRunner',
'--streaming',
'--region=us-central1'
]
with beam.Pipeline(options=PipelineOptions(pipeline_properties)) as p:
_ = (
p
| beam.io.ReadFromPubSub(
subscription=
'projects/dummy-project/subscriptions/dummy-subscription')
| beam.ExternalTransform(
'beam:transforms:xlang:test:prefix',
ImplicitSchemaPayloadBuilder({'data': '0'}),
expansion_service.ExpansionServiceServicer()))
pipeline_proto, _ = p.to_runner_api(return_context=True)
pubsub_read_transform = None
external_transform = None
proto_transforms = pipeline_proto.components.transforms
for id in proto_transforms:
if 'beam:transforms:xlang:test:prefix' in proto_transforms[
id].unique_name:
external_transform = proto_transforms[id]
if 'ReadFromPubSub' in proto_transforms[id].unique_name:
pubsub_read_transform = proto_transforms[id]
if not (pubsub_read_transform and external_transform):
raise ValueError(
'Could not find an external transform and the PubSub read transform '
'in the pipeline')
self.assertEqual(1, len(list(pubsub_read_transform.outputs.values())))
> self.assertEqual(
list(pubsub_read_transform.outputs.values()),
list(external_transform.inputs.values()))
E AssertionError: Lists differ: ['ref_PCollection_PCollection_2'] != ['ref_PCollection_PCollection_1']
E
E First differing element 0:
E 'ref_PCollection_PCollection_2'
E 'ref_PCollection_PCollection_1'
E
E - ['ref_PCollection_PCollection_2']
E ? ^
E
E + ['ref_PCollection_PCollection_1']
E ? ^
apache_beam/transforms/external_test.py:250: AssertionError
```
--
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] optimize grpc settings [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2481724839
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
+ # Default: 2, set to 0 to allow unlimited ping strikes
+ ("grpc.http2.max_ping_strikes", 0),
+ # Default: 0 (disabled), enable socket reuse for better handling
+ ("grpc.so_reuseport", 1),
Review Comment:
For prism runner, if the job_port pipeline option is not specified, its default value is 0:
https://github.com/apache/beam/blob/eba04b2a56759e6095b3cc9080c9941302543c57/sdks/python/apache_beam/options/pipeline_options.py#L1906-L1908
This port number is used by prism to call `pick_port`:
https://github.com/apache/beam/blob/eba04b2a56759e6095b3cc9080c9941302543c57/sdks/python/apache_beam/runners/portability/prism_runner.py#L486
Inside `pick_port`, we will first bind socket to 0 and then retrieve the port number if the port argument is 0.
https://github.com/apache/beam/blob/eba04b2a56759e6095b3cc9080c9941302543c57/sdks/python/apache_beam/utils/subprocess_server.py#L606-L607
This is exactly the case where this grpc option is useful.
--
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] optimize grpc settings [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2482116758
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
> Also note that because this value is now larger than grpc.keepalive_time_ms, it'll result in 3 outstanding keepalive pings,
per [comment above](https://github.com/apache/beam/pull/36528/files#r2482014530), keepalive_time_ms is INTMAX.
Increasing from 5 min to 10 min shouldn't be a problem - we don't expect dead channels, and if there was one, failing earlier likely won't help unless a system can somehow recover from a dead channel; yet, if the channel is not dead but a system is overloaded and a process is not responding, and increasing the TTL helps reduce flakiness, then I think it's fine to increase.
--
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] optimize grpc settings [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2499696403
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
Keep this with 5 min for now. We can explore this later if the tests can be improved with a longer time.
--
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] optimize grpc settings [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2481627059
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
Review Comment:
Thanks.
--
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] optimize grpc settings and fix some flaky tests [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on PR #36528:
URL: https://github.com/apache/beam/pull/36528#issuecomment-3426776969
https://github.com/apache/beam/pull/36579 to only fix the flaky tests
--
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] [TEST] increase grpc keepalive timeout and adjust ping settings [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #36528:
URL: https://github.com/apache/beam/pull/36528#issuecomment-3408577918
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/36528?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 36.21%. Comparing base ([`50f578a`](https://app.codecov.io/gh/apache/beam/commit/50f578aee326b20a04c09d8e82739f587893b24c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`c617972`](https://app.codecov.io/gh/apache/beam/commit/c61797241d3005b4d91c1918d30a0322c1d8035a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 13 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #36528 +/- ##
=============================================
- Coverage 55.08% 36.21% -18.87%
Complexity 1667 1667
=============================================
Files 1059 1059
Lines 165366 165376 +10
Branches 1195 1195
=============================================
- Hits 91091 59891 -31200
- Misses 72104 103314 +31210
Partials 2171 2171
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/36528/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [python](https://app.codecov.io/gh/apache/beam/pull/36528/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `40.51% <100.00%> (-40.50%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/beam/pull/36528?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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] optimize grpc settings [beam]
Posted by "sergiitk (via GitHub)" <gi...@apache.org>.
sergiitk commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2501689245
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
As discussed in another thread, this should be ok for your usage.
Note that without setting `grpc.keepalive_time_ms` in server channel ars, the server will send a keepalive ping every 2 hours.
So in the current setup, the server sends a ping every two hours, then waits for 10 minutes for client to return the ping
--
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] [TEST] increase grpc keepalive timeout and adjust ping settings [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #36528:
URL: https://github.com/apache/beam/pull/36528#issuecomment-3407689553
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
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] optimize grpc settings [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2449048203
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
Review Comment:
Would it make sense to use the same constant like `DEFAULT_OPTIONS` to define GRPC server settings in both places?
--
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] optimize grpc settings [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #36528:
URL: https://github.com/apache/beam/pull/36528#issuecomment-3427708507
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
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] optimize grpc settings [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2491978702
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
Do we have any periodic messages sent from SDK to runner that would otherwise detect a dead channel?
What if Dataflow runner crashes and restarts while the SDK had no active bundles. It seems like if we are not sending messages we might take up this this timeout to detect that the runner process we were talking to died and that we need to create a new channel.
If so this seems bad for streaming pipelines which want lower latency.
If this is just for testing, what about making this timeout configurable via an option?
--
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] optimize grpc settings [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #36528:
URL: https://github.com/apache/beam/pull/36528#issuecomment-3488303229
> Do we have any periodic messages sent from SDK to runner that would otherwise detect a dead channel?
I tried launching a pipeline, using an SDK with @liferoad 's changes patched, SSHing to the VM and restarting the 'harness' container to simulate the crash. SDK detected `Socket closed` error, and restarted within a few seconds. Logs:
```
DEFAULT 2025-11-04T22:54:52.114094879Z raise self
DEFAULT 2025-11-04T22:54:52.114100315Z grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
DEFAULT 2025-11-04T22:54:52.114105833Z status = StatusCode.UNAVAILABLE
DEFAULT 2025-11-04T22:54:52.114111214Z details = "Socket closed"
DEFAULT 2025-11-04T22:54:52.114119261Z debug_error_string = "UNKNOWN:Error received from peer ipv6:%5B::1%5D:12371 {created_time:"2025-11-04T22:54:51.032035467+00:00", grpc_status:14, grpc_message:"Socket closed"}"
DEFAULT 2025-11-04T22:54:52.114145002Z >
DEFAULT 2025-11-04T22:54:52.114150544Z {"stream":"stderr"}
DEFAULT 2025-11-04T22:54:52.114155948Z 2025/11/04 22:54:52 boot.go: error logging message over FnAPI. endpoint localhost:12370 error: EOF message follows
DEFAULT 2025-11-04T22:54:52.114161451Z 2025/11/04 22:54:52 WARN Python (worker sdk-0-0_sibling_1) exited 2 times: exit status 1
DEFAULT 2025-11-04T22:54:52.114167074Z restarting SDK process
```
--
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] [TEST] optimize grpc settings and fix some flaky tests [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2441206501
##########
sdks/python/apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:
##########
@@ -126,7 +126,7 @@ def test_enrichment_with_bigtable(self, mock_stdout):
enrichment_with_bigtable()
output = mock_stdout.getvalue().splitlines()
expected = validate_enrichment_with_bigtable()
- self.assertEqual(output, expected)
+ self.assertEqual(sorted(output), sorted(expected))
Review Comment:
```
=================================== FAILURES ===================================
_________________ EnrichmentTest.test_enrichment_with_bigtable _________________
[gw2] linux -- Python 3.10.17 /runner/_work/beam/beam/sdks/python/test-suites/tox/py310/build/srcs/sdks/python/target/.tox-py310-cloud/py310-cloud/bin/python
self = <apache_beam.examples.snippets.transforms.elementwise.enrichment_test.EnrichmentTest testMethod=test_enrichment_with_bigtable>
mock_stdout = <_io.StringIO object at 0x7bd7a443ef80>
def test_enrichment_with_bigtable(self, mock_stdout):
enrichment_with_bigtable()
output = mock_stdout.getvalue().splitlines()
expected = validate_enrichment_with_bigtable()
> self.assertEqual(output, expected)
E AssertionError: Lists differ: ["Row[141 chars]e_id=5, customer_id=5, product_id=4, quantity=[215 chars]'})"] != ["Row[141 chars]e_id=3, customer_id=3, product_id=2, quantity=[215 chars]'})"]
E
E First differing element 1:
E "Row(sale_id=5, customer_id=5, product_id=4, quantity=[77 chars]0'})"
E "Row(sale_id=3, customer_id=3, product_id=2, quantity=[76 chars]4'})"
E
E Diff is 650 characters long. Set self.maxDiff to None to see it.
apache_beam/examples/snippets/transforms/elementwise/enrichment_test.py:129: AssertionError
```
--
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] optimize grpc settings [beam]
Posted by "sergiitk (via GitHub)" <gi...@apache.org>.
sergiitk commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2479611499
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
Same logic as in https://github.com/apache/beam/pull/36528#discussion_r2479659838 applies here.
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
Review Comment:
The name should be `grpc.http2.min_ping_interval_without_data_ms`: https://github.com/grpc/grpc/blob/caaa581253b963b2ef3fb5b409254af1a657518a/include/grpc/impl/channel_arg_names.h#L112-L116
```c
/** Minimum allowed time between a server receiving successive ping frames
without sending any data/header frame. Int valued, milliseconds
*/
#define GRPC_ARG_HTTP2_MIN_RECV_PING_INTERVAL_WITHOUT_DATA_MS \
"grpc.http2.min_ping_interval_without_data_ms"
```
However, I don't think you even need that. See
> `RPC_ARG_HTTP2_MIN_RECV_PING_INTERVAL_WITHOUT_DATA_MS`
> If there are no data/header frames being sent on the transport, this channel argument on the server side controls the minimum time (in milliseconds) that gRPC Core would expect between receiving successive pings. If the time between successive pings is less that than this time, then the ping will be considered a bad ping from the peer. Such a ping counts as a ‘ping strike’. On the client side, this does not have any effect.
> — https://github.com/grpc/grpc/blob/master/doc/keepalive.md
Later you allow unlimited ping strikes.
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
Review Comment:
This option existed with a different name, but has been deprecated and now has no effect:
```c
/** (DEPRECATED) Does not have any effect.
Earlier, this arg configured the minimum time between successive ping frames
without receiving any data/header frame, Int valued, milliseconds. This put
unnecessary constraints on the configuration of keepalive pings,
requiring users to set this channel arg along with
GRPC_ARG_KEEPALIVE_TIME_MS. This arg also limited the activity of the other
source of pings in gRPC Core - BDP pings, but BDP pings are only sent when
there is receive-side data activity, making this arg unuseful for BDP pings
too. */
#define GRPC_ARG_HTTP2_MIN_SENT_PING_INTERVAL_WITHOUT_DATA_MS \
"grpc.http2.min_time_between_pings_ms"
```
https://github.com/grpc/grpc/blob/caaa581253b963b2ef3fb5b409254af1a657518a/include/grpc/impl/channel_arg_names.h#L101-L111
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
+ # Default: 2, set to 0 to allow unlimited ping strikes
+ ("grpc.http2.max_ping_strikes", 0),
Review Comment:
Seems reasonable
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
+ # Default: 2, set to 0 to allow unlimited ping strikes
+ ("grpc.http2.max_ping_strikes", 0),
+ # Default: 0 (disabled), enable socket reuse for better handling
+ ("grpc.so_reuseport", 1),
Review Comment:
It's not going to do anything usefull unless you specifically bind multiple server listeners to the same port. See https://github.com/grpc/grpc/tree/master/examples/python/multiprocessing#calculating-prime-numbers-with-multiple-processes.
This also can be useful when you want to bind a socket to a random (`:0`) port, get the actual port number, and only then bind the server to it.
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
Review Comment:
Seems reasonable. https://github.com/grpc/grpc/blob/master/doc/keepalive.md implying `0` is should've been the default in the first place:
> `GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA`
> This channel argument controls the maximum number of pings that can be sent when there is no data/header frame to be sent. gRPC Core will not continue sending pings if we run over the limit. Setting it to 0 allows sending pings without such a restriction.
> (Note that this is an unfortunate setting that does not agree with [A8-client-side-keepalive.md](https://github.com/grpc/proposal/blob/master/A8-client-side-keepalive.md). There should ideally be no such restriction on the keepalive ping and we plan to deprecate it in the future.)
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
+ # Default: 2, set to 0 to allow unlimited ping strikes
+ ("grpc.http2.max_ping_strikes", 0),
+ # Default: 0 (disabled), enable socket reuse for better handling
+ ("grpc.so_reuseport", 1),
+ # Default: 0 (disabled), set 30s TCP timeout for connection control
+ ("grpc.tcp_user_timeout_ms", 30000),
Review Comment:
Same as on the channel side, I do not believe this option exists.
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
Review Comment:
The default is 20 seconds
Channel Argument| Client|Server
----------------|-------|------
GRPC_ARG_KEEPALIVE_TIMEOUT_MS|20000 (20 seconds)|20000 (20 seconds)
https://github.com/grpc/grpc/blob/master/doc/keepalive.md
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
Review Comment:
Same as on the channel side, see https://github.com/apache/beam/pull/36528#discussion_r2479586543
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
Review Comment:
I recommend keeping these separate.
1. There are server-side only options
2. You don't always want to tune both client and the server exactly the same way
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
+ # Default: 2, set to 0 to allow unlimited ping strikes
+ ("grpc.http2.max_ping_strikes", 0),
Review Comment:
Yes, server-only.
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
Review Comment:
According to https://github.com/grpc/grpc/blob/master/doc/keepalive.md, the default on the server side is 2 hours:
Channel Argument| Client|Server
----------------|-------|------
GRPC_ARG_KEEPALIVE_TIME_MS|INT_MAX (disabled)|7200000 (2 hours)
Also see https://github.com/grpc/grpc/blob/caaa581253b963b2ef3fb5b409254af1a657518a/include/grpc/impl/channel_arg_names.h#L150-L153
```c
/** After a duration of this time the client/server pings its peer to see if the
transport is still alive. Int valued, milliseconds. Defaults to 7200000 (2
hours). */
#define GRPC_ARG_KEEPALIVE_TIME_MS "grpc.keepalive_time_ms"
```
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
This worries me. Consider what this option does:
> `GRPC_ARG_KEEPALIVE_TIMEOUT_MS`
> This channel argument controls the amount of time (in milliseconds) the sender of the keepalive ping waits for an acknowledgement. If it does not receive an acknowledgment within this time, it will close the connection.
> — https://github.com/grpc/grpc/blob/master/doc/keepalive.md
```c
/** After waiting for a duration of this time, if the keepalive ping sender does
not receive the ping ack, it will close the transport. Int valued,
milliseconds. Defaults to 20000 (20 seonds). */
#define GRPC_ARG_KEEPALIVE_TIMEOUT_MS "grpc.keepalive_timeout_ms"
```
> — https://github.com/grpc/grpc/blob/caaa581253b963b2ef3fb5b409254af1a657518a/include/grpc/impl/channel_arg_names.h#L154-L157
Are you really comfortable waiting 10 minutes until a dead channel is declared dead?
Also note that because this value is now larger than `grpc.keepalive_time_ms`, it'll result in 3 outstanding keepalive pings, while you're still waiting on the response for the first one.
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
+ # Default: 2, set to 0 to allow unlimited ping strikes
+ ("grpc.http2.max_ping_strikes", 0),
+ # Default: 0 (disabled), enable socket reuse for better handling
+ ("grpc.so_reuseport", 1),
+ # Default: 0 (disabled), set 30s TCP timeout for connection control
+ ("grpc.tcp_user_timeout_ms", 30000),
Review Comment:
I don't think this option even exists. See https://github.com/grpc/proposal/blob/master/A18-tcp-user-timeout.md.
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
Review Comment:
According to https://github.com/grpc/grpc/blob/master/doc/keepalive.md, there's no default value for this on the client side because by default the feature is disabled.
Channel Argument| Client|Server
----------------|-------|------
GRPC_ARG_KEEPALIVE_TIME_MS|INT_MAX (disabled)|7200000 (2 hours)
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
+ # Default: 2, set to 0 to allow unlimited ping strikes
+ ("grpc.http2.max_ping_strikes", 0),
+ # Default: 0 (disabled), enable socket reuse for better handling
+ ("grpc.so_reuseport", 1),
Review Comment:
Agree, this is server only, allowing it to bind sockets to the same port.
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
Review Comment:
Yes, this is server-side.
--
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] optimize grpc settings [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2482116758
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
> Also note that because this value is now larger than grpc.keepalive_time_ms, it'll result in 3 outstanding keepalive pings,
per [comment above](https://github.com/apache/beam/pull/36528/files#r2482014530), keepalive_time_ms is INTMAX. Is that correct?
Increasing from 5 min to 10 min shouldn't be a problem - we don't expect dead channels, and if there was one, failing earlier likely won't help unless a system can somehow recover from a dead channel; yet, if the channel is not dead but a system is overloaded and a process is not responding, and increasing the TTL helps reduce flakiness, then I think it's fine to increase.
--
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] optimize grpc settings [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2503785893
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,20 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
Review Comment:
Missed that. Fixed the 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] [TEST] increase grpc keepalive timeout and adjust ping settings [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #36528:
URL: https://github.com/apache/beam/pull/36528#issuecomment-3407303412
## Summary of Changes
Hello @liferoad, 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 optimizes GRPC channel settings within the Apache Beam Python SDK. The primary goal is to enhance performance and maintain connection stability by reducing the frequency of keepalive pings and adjusting various ping-related parameters. These changes are expected to minimize network overhead without compromising the reliability of GRPC communication.
### Highlights
* **GRPC Keepalive Time Adjustment**: The `grpc.keepalive_time_ms` setting has been increased from 20 seconds to 30 seconds. This change aims to reduce the frequency of keepalive pings, thereby decreasing unnecessary network traffic.
* **GRPC Ping Behavior Configuration**: New GRPC options have been introduced to fine-tune ping behavior. These include allowing unlimited pings without data (`grpc.http2.max_pings_without_data` set to 0), permitting keepalive pings even when no active calls are present (`grpc.keepalive_permit_without_calls` set to True), and establishing minimum intervals for receiving and sending pings without data (`grpc.http2.min_recv_ping_interval_without_data_ms` to 5 minutes and `grpc.http2.min_sent_ping_interval_without_data_ms` to 10 seconds).
<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 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 counter productive. 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).
<b>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](https://cloud.google.com/products/gemini/code-assist).</b>
</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] [TEST] increase grpc keepalive timeout and adjust ping settings [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #36528:
URL: https://github.com/apache/beam/pull/36528#issuecomment-3411076829
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] optimize grpc settings [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2481623502
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
Review Comment:
Updated the comment.
--
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] optimize grpc settings [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2482116758
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
> Also note that because this value is now larger than grpc.keepalive_time_ms, it'll result in 3 outstanding keepalive pings,
per [comment above](https://github.com/apache/beam/pull/36528/files#r2482014530), keepalive_time_ms is INTMAX. Is that correct, @sergiitk . Then this concern doesn't apply?
Increasing from 5 min to 10 min shouldn't be a problem - we don't expect dead channels, and if there was one, failing earlier likely won't help unless a system can somehow recover from a dead channel; yet, if the channel is not dead but a system is overloaded and a process is not responding, and increasing the TTL helps reduce flakiness, then I think it's fine to increase.
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
> Also note that because this value is now larger than grpc.keepalive_time_ms, it'll result in 3 outstanding keepalive pings,
per [comment above](https://github.com/apache/beam/pull/36528/files#r2482014530), keepalive_time_ms is INTMAX. Is that correct, @sergiitk ? Then this concern doesn't apply?
Increasing from 5 min to 10 min shouldn't be a problem - we don't expect dead channels, and if there was one, failing earlier likely won't help unless a system can somehow recover from a dead channel; yet, if the channel is not dead but a system is overloaded and a process is not responding, and increasing the TTL helps reduce flakiness, then I think it's fine to increase.
--
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] optimize grpc settings [beam]
Posted by "sergiitk (via GitHub)" <gi...@apache.org>.
sergiitk commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2501689245
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
As discussed in another thread, this should be ok for your usage.
Note that without setting `grpc.keepalive_time_ms`, the server will send a keepalive ping every 2 hours.
So in the current setup, the server sends a ping every two hours, then waits for 10 minutes for client to return the ping
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,28 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
+ # Default: 2, set to 0 to allow unlimited pings without data
+ ("grpc.http2.max_pings_without_data", 0),
+ # Default: False, set to True to allow keepalive pings when no calls
+ ("grpc.keepalive_permit_without_calls", True),
+ # Default: 300000ms (5min), increased to 10min for stability
+ ("grpc.http2.min_recv_ping_interval_without_data_ms", 600000),
+ # Default: 300000ms (5min), increased to 120s for conservative pings
+ ("grpc.http2.min_sent_ping_interval_without_data_ms", 120000),
+ # Default: 2, set to 0 to allow unlimited ping strikes
+ ("grpc.http2.max_ping_strikes", 0),
+ # Default: 0 (disabled), enable socket reuse for better handling
+ ("grpc.so_reuseport", 1),
Review Comment:
Great! With this option, you don't need to close the socket for the found port anymore, as you'll be able to bind and serve on it: https://github.com/apache/beam/blob/eba04b2a56759e6095b3cc9080c9941302543c57/sdks/python/apache_beam/utils/subprocess_server.py#L610-L612
You'll need to bind the initial socket with `SO_REUSEPORT`, ideally with `SO_REUSEADDR` as well.
https://github.com/apache/beam/blob/eba04b2a56759e6095b3cc9080c9941302543c57/sdks/python/apache_beam/utils/subprocess_server.py#L595
This approach addresses the race condition where an unused port was found by one process, closes the socket, but before this process starts listening on the found port, it's acquired by another process, resulting in `EADDRINUSE`.
By not closing the socket until the server stops listening, you'll prevent other processes from seeing that port as unused.
Note that this only applies to systems where `SO_REUSEPORT` is supported.
##########
sdks/python/apache_beam/utils/subprocess_server.py:
##########
@@ -185,8 +185,20 @@ def start(self):
try:
process, endpoint = self.start_process()
wait_secs = .1
- channel_options = [("grpc.max_receive_message_length", -1),
- ("grpc.max_send_message_length", -1)]
+ channel_options = [
+ ("grpc.max_receive_message_length", -1),
+ ("grpc.max_send_message_length", -1),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
Review Comment:
The default is 20000ms (20 seconds)
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,12 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 20000ms (20s), increased to 10 minutes for stability
Review Comment:
It's disabled by default, but unless you enable it, all other settings you added (`grpc.http2.max_pings_without_data`, `grpc.keepalive_permit_without_calls`) will have no effect.
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,12 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
+ # keep 5 minutes for now.
("grpc.keepalive_timeout_ms", 300000),
Review Comment:
Again, you want to keep `grpc.keepalive_time_ms`, but carefully consider the value for `grpc.keepalive_timeout_ms`
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
+ ("grpc.keepalive_time_ms", 180000),
+ # Default: 5000ms (5s), increased to 10 minutes for stability
+ ("grpc.keepalive_timeout_ms", 600000),
Review Comment:
@tvalentyn
> per [comment above](https://github.com/apache/beam/pull/36528/files#r2482014530), keepalive_time_ms is INTMAX. Is that correct, @sergiitk ? Then this concern doesn't apply?
`grpc.keepalive_time_ms = INTMAX` disables keep alive pings. Meaning `keepalive_timeout_ms` won't apply too, nor any of the keepalive settings.
--
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] optimize grpc settings and fix some flaky tests [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2445091439
##########
sdks/python/apache_beam/io/filebasedsink.py:
##########
@@ -205,6 +205,19 @@ def open_writer(self, init_result, uid):
# We also ensure there will be no collisions with uid and a
# (possibly unsharded) file_path_prefix and a (possibly empty)
# file_name_suffix.
+ from apache_beam.pvalue import EmptySideInput
Review Comment:
cc @shunping
--
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] optimize grpc settings and fix some flaky tests [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #36528:
URL: https://github.com/apache/beam/pull/36528#issuecomment-3423893469
i'd suggest to split grpc tune up into a dedicated change from test robustness changes.
--
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] optimize grpc settings [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #36528:
URL: https://github.com/apache/beam/pull/36528#issuecomment-3490061481
> > Do we have any periodic messages sent from SDK to runner that would otherwise detect a dead channel?
>
> I tried launching a pipeline, using an SDK with @liferoad 's changes patched, SSHing to the VM and restarting the 'harness' container to simulate the crash. SDK detected `Socket closed` error, and restarted within a few seconds. Logs:
>
> ```
> NOTICE 2025-11-04T22:54:50.975484Z valentyn : TTY=pts/1 ; PWD=/home/valentyn ; USER=root ; COMMAND=/var/lib/toolbox/nerdctl -n k8s.io restart 4a25ec1329e0
> ...
> DEFAULT 2025-11-04T22:54:52.114094879Z raise self
> DEFAULT 2025-11-04T22:54:52.114100315Z grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
> DEFAULT 2025-11-04T22:54:52.114105833Z status = StatusCode.UNAVAILABLE
> DEFAULT 2025-11-04T22:54:52.114111214Z details = "Socket closed"
> DEFAULT 2025-11-04T22:54:52.114119261Z debug_error_string = "UNKNOWN:Error received from peer ipv6:%5B::1%5D:12371 {created_time:"2025-11-04T22:54:51.032035467+00:00", grpc_status:14, grpc_message:"Socket closed"}"
> DEFAULT 2025-11-04T22:54:52.114145002Z >
> DEFAULT 2025-11-04T22:54:52.114150544Z {"stream":"stderr"}
> DEFAULT 2025-11-04T22:54:52.114155948Z 2025/11/04 22:54:52 boot.go: error logging message over FnAPI. endpoint localhost:12370 error: EOF message follows
> DEFAULT 2025-11-04T22:54:52.114161451Z 2025/11/04 22:54:52 WARN Python (worker sdk-0-0_sibling_1) exited 2 times: exit status 1
> DEFAULT 2025-11-04T22:54:52.114167074Z restarting SDK process
> ...
> INFO 2025-11-04T22:55:08.835663318Z Python sdk harness starting.
> ...
> INFO 2025-11-04T22:55:10.050536Z All SDK Harnesses registered!
> ```
Thanks Valentyn. Can we clarify the motivation for this in the PR better? If it is just perceived overhead of heartbeats, I can't imagine it is much and this doesn't seem worth risk of adding that additional latency in some cases. If it is to resolve unnecessary failures when we're CPU pegged that seems like better motivation and given the testing seems safe enough to try.
--
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] optimize grpc settings [beam]
Posted by "sergiitk (via GitHub)" <gi...@apache.org>.
sergiitk commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2501658416
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,12 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 20000ms (20s), increased to 10 minutes for stability
Review Comment:
It's disabled by default, and unless you enable it, all other settings you added (`grpc.http2.max_pings_without_data`, `grpc.keepalive_permit_without_calls`) will have no effect.
--
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] optimize grpc settings [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn merged PR #36528:
URL: https://github.com/apache/beam/pull/36528
--
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] optimize grpc settings [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2503816823
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,12 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 20000ms (20s), increased to 10 minutes for stability
Review Comment:
I see. I just added keepalive_time_ms back as before. Since both values are same as before, I think it should be fine for our case.
--
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] optimize grpc settings and fix some flaky tests [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2445044742
##########
sdks/python/apache_beam/io/filebasedsink.py:
##########
@@ -205,6 +205,19 @@ def open_writer(self, init_result, uid):
# We also ensure there will be no collisions with uid and a
# (possibly unsharded) file_path_prefix and a (possibly empty)
# file_name_suffix.
+ from apache_beam.pvalue import EmptySideInput
Review Comment:
We can do this, but I think it represents a bug with prism's handling of side inputs. In this case, https://github.com/apache/beam/blob/af748d07a1a7ca244ff057554d68932c014d5eeb/sdks/python/apache_beam/io/iobase.py#L1173 hasn't executed, which means that `sink.initialize_write()` hasn't necessarily been called yet, but this should be guaranteed for batch pipelines
So we should probably file a follow up issue
--
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] optimize grpc settings [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on PR #36528:
URL: https://github.com/apache/beam/pull/36528#issuecomment-3427694268
R: @sergiitk
--
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] optimize grpc settings [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #36528:
URL: https://github.com/apache/beam/pull/36528#discussion_r2481621820
##########
sdks/python/apache_beam/runners/worker/channel_factory.py:
##########
@@ -23,8 +23,24 @@
class GRPCChannelFactory(grpc.StreamStreamClientInterceptor):
DEFAULT_OPTIONS = [
- ("grpc.keepalive_time_ms", 20000),
- ("grpc.keepalive_timeout_ms", 300000),
+ # Default: 30000ms (30s), increased to 180s to reduce ping frequency
Review Comment:
Removed this for now.
--
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