feat: support configuring the client keep-alive interval

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)

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)

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))