Fix ssl timeout errors for Quix applications (#254)

Author: daniil-quix

quixstreams/app.py

@@ -164,7 +164,6 @@ def __init__(
         self._on_processing_error = on_processing_error or default_on_processing_error
         self._on_message_processed = on_message_processed
         self._quix_config_builder: Optional[QuixKafkaConfigsBuilder] = None
-        self._state_manager: Optional[StateStoreManager] = None
         self._state_manager = StateStoreManager(
             group_id=consumer_group,
             state_dir=state_dir,

quixstreams/platforms/quix/config.py

@@ -19,6 +19,9 @@
     "TopicCreationConfigs",
 )
 
+QUIX_CONNECTIONS_MAX_IDLE_MS = 3 * 60 * 1000
+QUIX_METADATA_MAX_AGE_MS = 3 * 60 * 1000
+
 
 @dataclasses.dataclass
 class TopicCreationConfigs:
@@ -410,6 +413,18 @@ def get_confluent_broker_config(self, known_topic: Optional[str] = None) -> dict
             cfg_out[c_cfg] = self.QuixApiKafkaAuthConfigMap.values.get(value, value)
         cfg_out["ssl.endpoint.identification.algorithm"] = "none"
         cfg_out["ssl.ca.location"] = self._set_workspace_cert()
+
+        # Set the connection idle timeout and metadata max age to be less than
+        # Azure's default 4 minutes.
+        # Azure LB kills the inbound TCP connections after 4 mins and these settings
+        # help to handle that.
+        # More about this issue:
+        # - https://github.com/confluentinc/librdkafka/issues/3109
+        # - https://learn.microsoft.com/en-us/azure/event-hubs/apache-kafka-configurations#producer-and-consumer-configurations-1
+        # These values can be overwritten on the Application level by passing
+        # `extra_consumer_config` or `extra_producer_config` parameters.
+        cfg_out["connections.max.idle.ms"] = QUIX_CONNECTIONS_MAX_IDLE_MS
+        cfg_out["metadata.max.age.ms"] = QUIX_METADATA_MAX_AGE_MS
         self._confluent_broker_config = cfg_out
         return self._confluent_broker_config
 

tests/test_quixstreams/test_platforms/test_quix/test_config.py

@@ -8,6 +8,10 @@
 from requests import HTTPError, Response
 
 from quixstreams.models.topics import Topic
+from quixstreams.platforms.quix.config import (
+    QUIX_CONNECTIONS_MAX_IDLE_MS,
+    QUIX_METADATA_MAX_AGE_MS,
+)
 
 
 class TestQuixKafkaConfigsBuilder:
@@ -404,6 +408,8 @@ def test_get_confluent_broker_config(self, quix_kafka_config_factory):
             "sasl.password": "my-password",
             "ssl.ca.location": "/mock/dir/ca.cert",
             "ssl.endpoint.identification.algorithm": "none",
+            "connections.max.idle.ms": QUIX_CONNECTIONS_MAX_IDLE_MS,
+            "metadata.max.age.ms": QUIX_METADATA_MAX_AGE_MS,
         }
 
     def test_append_workspace_id(self, quix_kafka_config_factory):