Readd Confluent Kafka transport driver (rebased on master, working on released confluent-kafka)

.coveragerc

@@ -23,6 +23,10 @@ omit =
     */faust/assignor/*
     */faust/transport/drivers/memory.py
 
+    # optional driver: only importable/tested with the ckafka extra installed
+    # (faust[ckafka]); not part of the CI test environment.
+    */faust/transport/drivers/confluent.py
+
     # tested by integration
     */faust/tables/recovery.py
 

docs/conf.py

@@ -54,6 +54,7 @@
         'faust.cli',
         'faust.models',
         'faust.serializers',
+        'faust.transport.drivers.confluent',
         'faust.types',
         'faust.types._env',
         'faust.utils',

docs/includes/settingref.txt

@@ -391,6 +391,15 @@ You can also pass a list of URLs:
    Limitations: None
 
 
+- ``confluent://``
+
+   Experimental transport using the :pypi:`confluent-kafka` client.
+
+   Limitations: Does not do sticky partition assignment (not
+       suitable for tables), and do not create any necessary internal
+       topics (you have to create them manually).
+
+
 .. setting:: broker_credentials
 
 ``broker_credentials``

extra/bandit/baseline.json

@@ -1070,6 +1070,18 @@
       "loc": 890,
       "nosec": 0
     },
+    "faust/transport/drivers/confluent.py": {
+      "CONFIDENCE.HIGH": 0.0,
+      "CONFIDENCE.LOW": 0.0,
+      "CONFIDENCE.MEDIUM": 0.0,
+      "CONFIDENCE.UNDEFINED": 0.0,
+      "SEVERITY.HIGH": 0.0,
+      "SEVERITY.LOW": 0.0,
+      "SEVERITY.MEDIUM": 0.0,
+      "SEVERITY.UNDEFINED": 0.0,
+      "loc": 486,
+      "nosec": 0
+    },
     "faust/transport/producer.py": {
       "CONFIDENCE.HIGH": 0.0,
       "CONFIDENCE.LOW": 0.0,
@@ -2043,4 +2055,4 @@
       "test_name": "blacklist"
     }
   ]
-}
+}

faust/transport/drivers/__init__.py

@@ -1,22 +1,18 @@
 """Transport registry."""
 
-from yarl import URL
+from typing import Type
 
-from .aiokafka import Transport as AIOKafkaTransport
+from mode.utils.imports import FactoryMapping
 
-__all__ = ["by_name", "by_url"]
-
-
-DRIVERS = {
-    "aiokafka": AIOKafkaTransport,
-    "kafka": AIOKafkaTransport,
-}
-
-
-def by_name(driver_name: str):
-    return DRIVERS[driver_name]
+from faust.types import TransportT
 
+__all__ = ["by_name", "by_url"]
 
-def by_url(url: URL):
-    scheme = url.scheme
-    return DRIVERS[scheme]
+TRANSPORTS: FactoryMapping[Type[TransportT]] = FactoryMapping(
+    aiokafka="faust.transport.drivers.aiokafka:Transport",
+    confluent="faust.transport.drivers.confluent:Transport",
+    kafka="faust.transport.drivers.aiokafka:Transport",
+)
+TRANSPORTS.include_setuptools_namespace("faust.transports")
+by_name = TRANSPORTS.by_name
+by_url = TRANSPORTS.by_url