From ca91685bd32fcc91659ad92c2350f0662f4de36e Mon Sep 17 00:00:00 2001 From: M-Schwab Date: Thu, 29 Oct 2026 08:45:57 -0700 Subject: [PATCH] feat: support configuring the client keep-alive interval Expose the C++ client's ClientConfiguration keep-alive interval as Client(keep_alive_interval_seconds=30). Generated-by: Claude Code 2.1.312 (Claude Opus 5.5) --- pulsar/__init__.py | 10 ++++++++++ src/config.cc | 3 +++ tests/pulsar_test.py | 2 ++ 3 files changed, 15 insertions(+) diff --git a/pulsar/__init__.py b/pulsar/__init__.py index c1600f2..3deeb42 100644 --- a/pulsar/__init__.py +++ b/pulsar/__init__.py @@ -856,6 +856,7 @@ def __init__(self, service_url, listener_name=None, tls_private_key_file_path: Optional[str] = None, tls_certificate_file_path: Optional[str] = None, + keep_alive_interval_seconds: int = 30, ): """ Create a new Pulsar client instance. @@ -925,6 +926,11 @@ def __init__(self, service_url, The path to the TLS private key file tls_certificate_file_path: str, optional The path to the TLS certificate file. + keep_alive_interval_seconds: int, default=30 + Set the interval in seconds between keep-alive pings on each broker connection. A connection + that has received nothing from the broker since its last ping when the next one is due is + closed and reconnected. Raise it on slow or congested links, where the broker's replies can + wait behind queued messages for longer than this. """ if not isinstance(service_url, (str, AutoClusterFailover, ServiceInfoProvider)): raise ValueError( @@ -961,12 +967,16 @@ def __init__(self, service_url, _check_type_or_none(str, listener_name, 'listener_name') _check_type_or_none(str, tls_private_key_file_path, 'tls_private_key_file_path') _check_type_or_none(str, tls_certificate_file_path, 'tls_certificate_file_path') + _check_type(int, keep_alive_interval_seconds, 'keep_alive_interval_seconds') + if keep_alive_interval_seconds <= 0: + raise ValueError("Argument keep_alive_interval_seconds is expected to be positive") conf = _pulsar.ClientConfiguration() if authentication: conf.authentication(authentication.auth) conf.operation_timeout_seconds(operation_timeout_seconds) conf.connection_timeout(connection_timeout_ms) + conf.keep_alive_interval_in_seconds(keep_alive_interval_seconds) conf.io_threads(io_threads) conf.message_listener_threads(message_listener_threads) conf.concurrent_lookup_requests(concurrent_lookup_requests) diff --git a/src/config.cc b/src/config.cc index ec16b52..ef7c810 100644 --- a/src/config.cc +++ b/src/config.cc @@ -184,6 +184,9 @@ void export_config(py::module_& m) { return_value_policy::reference) .def("connection_timeout", &ClientConfiguration::getConnectionTimeout) .def("connection_timeout", &ClientConfiguration::setConnectionTimeout, return_value_policy::reference) + .def("keep_alive_interval_in_seconds", &ClientConfiguration::getKeepAliveIntervalInSeconds) + .def("keep_alive_interval_in_seconds", &ClientConfiguration::setKeepAliveIntervalInSeconds, + return_value_policy::reference) .def("io_threads", &ClientConfiguration::getIOThreads) .def("io_threads", &ClientConfiguration::setIOThreads, return_value_policy::reference) .def("message_listener_threads", &ClientConfiguration::getMessageListenerThreads) diff --git a/tests/pulsar_test.py b/tests/pulsar_test.py index 68eb499..d4d5b68 100755 --- a/tests/pulsar_test.py +++ b/tests/pulsar_test.py @@ -964,6 +964,8 @@ def test_client_argument_errors(self): self._check_value_error(lambda: Client(self.serviceUrl, use_tls="test")) self._check_value_error(lambda: Client(self.serviceUrl, tls_trust_certs_file_path=5)) self._check_value_error(lambda: Client(self.serviceUrl, tls_allow_insecure_connection="test")) + self._check_value_error(lambda: Client(self.serviceUrl, keep_alive_interval_seconds="test")) + self._check_value_error(lambda: Client(self.serviceUrl, keep_alive_interval_seconds=0)) def test_service_info_argument_errors(self): self._check_value_error(lambda: ServiceInfo(None))