Updates for client_id and group_id

RasaHQ · Oct 16, 2020 · 1c26d58 · 1c26d58
1 parent 7369560
commit 1c26d58
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/rasa/core/brokers/kafka.py b/rasa/core/brokers/kafka.py
@@ -14,7 +14,6 @@ def __init__(
         url: Union[Text, List[Text], None],
         topic: Text = "rasa_core_events",
         client_id: Optional[Text] = None,
-        group_id: Optional[Text] = None,
         sasl_username: Optional[Text] = None,
         sasl_password: Optional[Text] = None,
         ssl_cafile: Optional[Text] = None,
@@ -63,7 +62,6 @@ def __init__(
         self.url = url
         self.topic = topic
         self.client_id = client_id
-        self.group_id = group_id
         self.security_protocol = security_protocol.upper()
         self.sasl_username = sasl_username
         self.sasl_password = sasl_password
@@ -93,13 +91,15 @@ def _create_producer(self) -> None:
 
         if self.security_protocol == "PLAINTEXT":
             self.producer = kafka.KafkaProducer(
+                client_id=self.client_id,
                 bootstrap_servers=self.url,
                 value_serializer=lambda v: json.dumps(v).encode(DEFAULT_ENCODING),
                 security_protocol=self.security_protocol,
                 ssl_check_hostname=False,
             )
         elif self.security_protocol == "SASL_PLAINTEXT":
             self.producer = kafka.KafkaProducer(
+                client_id=self.client_id,
                 bootstrap_servers=self.url,
                 value_serializer=lambda v: json.dumps(v).encode(DEFAULT_ENCODING),
                 sasl_plain_username=self.sasl_username,
@@ -109,6 +109,7 @@ def _create_producer(self) -> None:
             )
         elif self.security_protocol == "SSL":
             self.producer = kafka.KafkaProducer(
+                client_id=self.client_id,
                 bootstrap_servers=self.url,
                 value_serializer=lambda v: json.dumps(v).encode(DEFAULT_ENCODING),
                 ssl_cafile=self.ssl_cafile,
@@ -119,6 +120,7 @@ def _create_producer(self) -> None:
             )
         elif self.security_protocol == "SASL_SSL":
             self.producer = kafka.KafkaProducer(
+                client_id=self.client_id,
                 bootstrap_servers=self.url,
                 value_serializer=lambda v: json.dumps(v).encode(DEFAULT_ENCODING),
                 sasl_plain_username=self.sasl_username,