From 0edaeb12586dd639a026329e1ac0963b7ccb6fc8 Mon Sep 17 00:00:00 2001
From: magqqgq <146786427+magqqgq@users.noreply.github.com>
Date: Mon, 17 Aug 2026 01:35:42 +0300
Subject: [PATCH] Refactor: Replace silent capability stubs with explicit
 NotImplementedError failures

Why
The Confluent Kafka transport driver previously contained placeholder stubs for topic creation and partition lookup capabilities that failed silently or lacked explicit interface enforcement.

How
Replaced unreachable or silent topic/partition capability stubs with explicit NotImplementedError failures across the consumer and producer components.

Security/Robustness changes
Improves API contract clarity by raising explicit errors when unsupported transport operations are invoked.

Testing
Verified that the modified Python file compiles cleanly using py_compile.
---
 faust/transport/drivers/confluent.py | 38 +++++++++-------------------
 1 file changed, 12 insertions(+), 26 deletions(-)

diff --git a/faust/transport/drivers/confluent.py b/faust/transport/drivers/confluent.py
index f0e47816d..b5de1ab6a 100644
--- a/faust/transport/drivers/confluent.py
+++ b/faust/transport/drivers/confluent.py
@@ -82,7 +82,9 @@ async def create_topic(self,
                            deleting: bool = None,
                            ensure_created: bool = False) -> None:
         """Create topic on broker."""
-        return  # XXX
+        raise NotImplementedError(
+                'Topic creation is not supported by the Confluent consumer'
+            )
         await self._thread.create_topic(
             topic,
             partitions,
@@ -352,7 +354,9 @@ def key_partition(self,
                       topic: str,
                       key: Optional[bytes],
                       partition: int = None) -> Optional[int]:
-        raise NotImplementedError('TODO')  # TODO XXX
+        raise NotImplementedError(
+                'Partition lookup is not supported by the Confluent consumer'
+            )
 
 
 class ProducerProduceFuture(asyncio.Future):
@@ -478,23 +482,9 @@ async def create_topic(self,
                            deleting: bool = None,
                            ensure_created: bool = False) -> None:
         """Create topic on broker."""
-        return  # XXX
-        _retention = (int(want_seconds(retention) * 1000.0)
-                      if retention else None)
-        await cast(Transport, self.transport)._create_topic(
-            self,
-            self._producer.client,
-            topic,
-            partitions,
-            replication,
-            config=config,
-            timeout=int(want_seconds(timeout) * 1000.0),
-            retention=_retention,
-            compacting=compacting,
-            deleting=deleting,
-            ensure_created=ensure_created,
-        )
-
+        raise NotImplementedError(
+                'Topic creation is not supported by the Confluent producer'
+            )
     async def on_start(self) -> None:
         """Call when producer is starting."""
         await self._producer_thread.start()
@@ -519,12 +509,6 @@ async def send(self, topic: str, key: Optional[bytes],
             on_delivery=fut.set_from_on_delivery,
         )
         return cast(Awaitable[RecordMetadata], fut)
-        try:
-            return cast(Awaitable[RecordMetadata], await self._producer.send(
-                topic, value, key=key, partition=partition))
-        except KafkaException as exc:
-            raise ProducerSendError(f'Error while sending: {exc!r}') from exc
-
     async def send_and_wait(self, topic: str, key: Optional[bytes],
                             value: Optional[bytes],
                             partition: Optional[int],
@@ -548,7 +532,9 @@ async def flush(self) -> None:
 
     def key_partition(self, topic: str, key: bytes) -> TP:
         """Return topic and partition destination for key."""
-        raise NotImplementedError()
+        raise NotImplementedError(
+                'Partition lookup is not supported by the Confluent producer'
+            )
 
 
 class Transport(base.Transport):