From ad2588a424887feeebfe503fbd0b7548777635fc Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Fri, 5 Dec 2025 12:24:37 -0500 Subject: [PATCH 1/2] Fix too-many-pings on FnAPI runner under grpc mode --- .../runners/portability/fn_api_runner/worker_handlers.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py b/sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py index 338f6ece57c0..2acca7a4d942 100644 --- a/sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py +++ b/sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py @@ -468,7 +468,11 @@ def __init__( options = [("grpc.max_receive_message_length", -1), ("grpc.max_send_message_length", -1), ("grpc.http2.max_pings_without_data", 0), - ("grpc.http2.max_ping_strikes", 0)] + ("grpc.http2.max_ping_strikes", 0), + # match `grpc.keepalive_time_ms` defined in the client + # (channel_factory.py) + ("grpc.http2.min_ping_interval_without_data_ms", 20_000), + ] self.state = state self.provision_info = provision_info From 95c97347e7a847c4226d5fa35540ced572fd47f5 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Fri, 5 Dec 2025 12:43:02 -0500 Subject: [PATCH 2/2] Fix lints --- .../fn_api_runner/worker_handlers.py | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py b/sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py index 2acca7a4d942..d79b381f2d78 100644 --- a/sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py +++ b/sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py @@ -465,14 +465,15 @@ def __init__( # received or sent over the data plane. The actual buffer size # is controlled in a layer above. Also, options to keep the server alive # when too many pings are received. - options = [("grpc.max_receive_message_length", -1), - ("grpc.max_send_message_length", -1), - ("grpc.http2.max_pings_without_data", 0), - ("grpc.http2.max_ping_strikes", 0), - # match `grpc.keepalive_time_ms` defined in the client - # (channel_factory.py) - ("grpc.http2.min_ping_interval_without_data_ms", 20_000), - ] + options = [ + ("grpc.max_receive_message_length", -1), + ("grpc.max_send_message_length", -1), + ("grpc.http2.max_pings_without_data", 0), + ("grpc.http2.max_ping_strikes", 0), + # match `grpc.keepalive_time_ms` defined in the client + # (channel_factory.py) + ("grpc.http2.min_ping_interval_without_data_ms", 20_000), + ] self.state = state self.provision_info = provision_info