Add ping method to async confluent producer and fix confluent ping method

Author: kumaranvpl

faststream/confluent/broker/broker.py

@@ -533,3 +533,5 @@ async def ping(self, timeout: Optional[float]) -> bool:
 
             if self._producer is None:
                 return False
+
+            return await self._producer._producer.ping(timeout=timeout)

faststream/confluent/client.py

@@ -26,6 +26,19 @@
 
 _missing = object()
 
+ADMINCLIENT_CONFIG_PARAMS = (
+    "allow.auto.create.topics",
+    "bootstrap.servers",
+    "client.id",
+    "request.timeout.ms",
+    "metadata.max.age.ms",
+    "security.protocol",
+    "connections.max.idle.ms",
+    "sasl.mechanism",
+    "sasl.username",
+    "sasl.password",
+)
+
 
 class MsgToSend(BaseModel):
     """A Pydantic model representing a message to be sent to Kafka.
@@ -214,6 +227,24 @@ async def send_batch(
         ]
         await asyncio.gather(*tasks)
 
+    async def ping(self, timeout: Optional[float] = 5.0) -> bool:
+        """Implement ping using AdminClient."""
+        try:
+            admin_client = AdminClient(
+                {
+                    x: self.config[x]
+                    for x in ADMINCLIENT_CONFIG_PARAMS
+                    if x in self.config
+                }
+            )
+            cluster_metadata = await call_or_await(
+                admin_client.list_topics, timeout=timeout
+            )
+
+            return bool(cluster_metadata)
+        except Exception:
+            return False
+
 
 class TopicPartition(NamedTuple):
     """A named tuple representing a Kafka topic and partition."""
@@ -228,22 +259,8 @@ def create_topics(
     logger: Union["LoggerProto", None, object] = logger,
 ) -> None:
     """Creates Kafka topics using the provided configuration."""
-    required_config_params = (
-        "allow.auto.create.topics",
-        "bootstrap.servers",
-        "client.id",
-        "request.timeout.ms",
-        "metadata.max.age.ms",
-        "security.protocol",
-        "connections.max.idle.ms",
-        "sasl.mechanism",
-        "sasl.username",
-        "sasl.password",
-        "sasl.kerberos.service.name",
-    )
-
     admin_client = AdminClient(
-        {x: config[x] for x in required_config_params if x in config}
+        {x: config[x] for x in ADMINCLIENT_CONFIG_PARAMS if x in config}
     )
 
     fs = admin_client.create_topics(