fix kafka admin error (#16)

deepchecks · Apr 17, 2023 · b1efd86 · b1efd86
1 parent 267d3c2
commit b1efd86
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 1 deletion.
diff --git a/backend/deepchecks_monitoring/bgtasks/model_version_topic_delete.py b/backend/deepchecks_monitoring/bgtasks/model_version_topic_delete.py
@@ -48,7 +48,7 @@ async def run(self, task: 'Task', session: AsyncSession, resources_provider: Res
         if self.kafka_admin is None:
             with self.lock:
                 if self.kafka_admin is None:
-                    self.kafka_admin = AIOKafkaAdminClient(**resources_provider.kafka_settings.kafka_params)
+                    self.kafka_admin = AIOKafkaAdminClient(**resources_provider.kafka_settings.kafka_admin_params)
                     await self.kafka_admin.start()
 
         # Backward compatibility, remove in next release and replace with:

diff --git a/backend/deepchecks_monitoring/config.py b/backend/deepchecks_monitoring/config.py
@@ -79,6 +79,16 @@ def kafka_params(self):
             'metadata_max_age_ms': self.kafka_max_metadata_age
         }
 
+    @property
+    def kafka_admin_params(self):
+        """Get connection parameters for kafka admin."""
+        return {
+            'bootstrap_servers': self.kafka_host,
+            'security_protocol': self.kafka_security_protocol,
+            'ssl_context': create_ssl_context(),
+            'metadata_max_age_ms': self.kafka_max_metadata_age
+        }
+
 
 class DatabaseSettings(BaseDeepchecksSettings):
     """Database settings."""