Kafka cluster setup

README.rst

@@ -75,6 +75,38 @@ If you're using AWS Kinesis instead of Kafka, it will look like this:
         # 'PRODUCER_ID': 'my-application-name',
     }
 
+If you're need to set any other attributes, you cas use "KAFKA_KWARGS"
+
+For example with Confluent Cloud will look like this:
+
+::
+
+    LOGPIPE = {
+        ...
+        # Optional Settings
+        'KAFKA_KWARGS': {
+            'security_protocol': 'SASL_SSL',
+            'sasl_mechanism': 'PLAIN',
+            'sasl_plain_username': '<api_key>',
+            'sasl_plain_password': '<api_secret>',
+        }
+    }
+
+or with OVHcloud will look like this:
+
+::
+
+    LOGPIPE = {
+        ...
+        # Optional Settings
+        'KAFKA_KWARGS': {
+            'security_protocol': 'SSL',
+            'ssl_cafile': '<ca.pem or ca.certificate.pem>',
+            'ssl_certfile': '<service.cert or access.certificate.pem>',
+            'ssl_keyfile': '<service.key or access.key>',
+        }
+    }
+
 Run migrations. This will create the model used to store Kafka log position offsets.::
 
     $ python manage.py migrate logpipe

src/logpipe/backend/kafka.py

@@ -115,6 +115,7 @@ def _get_client_config(self):
             "enable_auto_commit": False,
             "consumer_timeout_ms": 1000,
         }
+        kwargs.update(settings.get("KAFKA_KWARGS", {}))
         kwargs.update(settings.get("KAFKA_CONSUMER_KWARGS", {}))
         kwargs.update(self.client_kwargs)
         kwargs.update(
@@ -145,9 +146,9 @@ def send(self, topic_name, key, value):
         )
 
     def _get_client_config(self):
-        servers = settings.get("KAFKA_BOOTSTRAP_SERVERS")
-        retries = settings.get("KAFKA_MAX_SEND_RETRIES", 0)
-        return {
-            "bootstrap_servers": servers,
-            "retries": retries,
+        kwargs = {
+            "bootstrap_servers": settings.get("KAFKA_BOOTSTRAP_SERVERS"),
+            "retries": settings.get("KAFKA_MAX_SEND_RETRIES", 0),
         }
+        kwargs.update(settings.get("KAFKA_KWARGS", {}))
+        return kwargs