wip

automation/devops_automation_infra/k8s_plugins/kafka.py

@@ -45,6 +45,7 @@ def _is_exposed(self):
 
     def _expose(self):
         # Checks if kafka is already exposed
+        import ipdb;ipdb.set_trace()
         if self._is_exposed:
             return
 
@@ -80,9 +81,12 @@ def _add_default_options(self, kwargs):
 
     def _bootstrap_endpoint(self):
         v1 = kubernetes.client.CoreV1Api(self._cluster.Kubectl.client())
-        port = v1.read_namespaced_service(namespace=self._namespace, name='kafka-cluster-kafka-external-bootstrap').spec.ports[0].node_port
+        port = v1.read_namespaced_service(namespace=self._namespace, name='kafka-cluster-kafka-external-bootstrap').spec.ports[0].port
         return f"{self._master.ip}:{port}"
 
+    def change_namespace(self,namespace):
+        self._namespace = namespace
+
     def admin(self, **kwargs):
         self._expose()
         options = self._add_default_options(kwargs)
@@ -91,7 +95,7 @@ def admin(self, **kwargs):
     def consumer(self, *topics, **kwargs):
         self._expose()
         options = self._add_default_options(kwargs)
-        return kafka.KafkaConsumer(*topics, **options)
+        return kafka.KafkaConsumer(security_protocol="SSL",*topics, **options)
 
     def producer(self, **kwargs):
         self._expose()

automation/proxy_container/requirements.txt

@@ -8,7 +8,7 @@ boto3
 grpcio==1.21.1
 PyMySQL==0.9.3
 python-consul
-kafka-python==2.0.1
+kafka-python==2.0.2
 munch==2.5.0
 pyzmq==19.0.1
 paho-mqtt==1.5.1