fix(ingest): kafka ingest task hand up with error bootstrap server (d…

…atahub-project#6820)

metadata-ingestion/src/datahub/configuration/kafka.py

@@ -16,6 +16,8 @@ class _KafkaConnectionConfig(ConfigModel):
         description="Extra schema registry config serialized as JSON. These options will be passed into Kafka's SchemaRegistryClient. https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html?#schemaregistryclient",
     )
 
+    client_timeout_seconds: int = Field(default=60, description="The request timeout used when interacting with the Kafka APIs.")
+
     @validator("bootstrap")
     def bootstrap_host_colon_port_comma(cls, val: str) -> str:
         for entry in val.split(","):

metadata-ingestion/src/datahub/ingestion/source/kafka.py

@@ -200,7 +200,9 @@ def create(cls, config_dict: Dict, ctx: PipelineContext) -> "KafkaSource":
         return cls(config, ctx)
 
     def get_workunits(self) -> Iterable[MetadataWorkUnit]:
-        topics = self.consumer.list_topics().topics
+        topics = self.consumer.list_topics(
+            timeout=self.source_config.connection.client_timeout_seconds
+        ).topics
         extra_topic_details = self.fetch_extra_topic_details(topics.keys())
 
         for t, t_detail in topics.items():