Configure the confluent:// transport from app settings via kafkaesq

docs/conf.py

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

docs/includes/installation.txt

@@ -57,6 +57,12 @@ Transports
     for using the :pypi:`confluent-kafka` client through the ``confluent://``
     transport.
 
+:``faust[kafkaesq]``:
+    for converting Kafka client configuration between the spellings
+    :pypi:`aiokafka`, :pypi:`confluent-kafka` and Faust app settings use --
+    turning an existing librdkafka config into app settings, or app settings
+    into arguments for a plain client. See :ref:`client-config-conversion`.
+
 Codecs
 ~~~~~~
 

docs/includes/settingref.txt

@@ -390,6 +390,8 @@ You can also pass a list of URLs:
 
    Limitations: None
 
+   Reference: :mod:`faust.transport.drivers.aiokafka`
+
 
 - ``confluent://``
 
@@ -399,6 +401,63 @@ You can also pass a list of URLs:
        suitable for tables), and do not create any necessary internal
        topics (you have to create them manually).
 
+   Reference: :mod:`faust.transport.drivers.confluent`
+
+.. _client-config-conversion:
+
+**Converting client configuration**
+
+The clients behind these transports spell their configuration
+differently: :pypi:`aiokafka` takes ``session_timeout_ms``, librdkafka
+takes ``session.timeout.ms``, and app settings are a third spelling
+again (:setting:`broker_session_timeout`, in seconds). Install the
+``kafkaesq`` bundle:
+
+.. sourcecode:: console
+
+   $ pip install "faust-streaming[kafkaesq]"
+
+and :class:`faust.transport.kafkaesq.ClientConfig` converts between
+them. A librdkafka config -- a Confluent Cloud ``client.properties``,
+say -- becomes arguments for :class:`faust.App`:
+
+.. sourcecode:: python
+
+   from faust.transport.kafkaesq import ClientConfig
+
+   config = ClientConfig.from_confluent({
+       'bootstrap.servers': 'kafka.example.com:9092',
+       'group.id': 'billing',
+       'session.timeout.ms': 45000,
+   })
+
+   settings = config.as_app_settings()
+   app = faust.App(settings.pop('id'), **settings)
+
+and, the other way round, an app's settings configure a plain client
+that talks to the same brokers:
+
+.. sourcecode:: python
+
+   from faust.transport.kafkaesq import PRODUCER_SETTINGS, ClientConfig
+
+   config = ClientConfig.from_app(app, PRODUCER_SETTINGS)
+   producer = confluent_kafka.Producer(config.as_confluent())
+   # ... or the same settings as aiokafka kwargs
+   producer = AIOKafkaProducer(**config.as_aiokafka())
+
+This is a conversion helper, not something the transports use: both
+configure themselves from your app settings as they always have, and
+neither needs :pypi:`kafkaesq` installed.
+
+.. note::
+
+   Authentication is not converted. Faust configures it with a
+   :setting:`broker_credentials` object built at runtime, which no
+   config file can describe, so ``security.protocol``, ``sasl.*``
+   and ``ssl.*`` keys are reported as unmapped when converting to
+   app settings -- set :setting:`broker_credentials` yourself.
+
 
 .. setting:: broker_credentials
 

docs/reference/faust.transport.drivers.confluent.rst

@@ -0,0 +1,11 @@
+=====================================================
+ ``faust.transport.drivers.confluent``
+=====================================================
+
+.. contents::
+    :local:
+.. currentmodule:: faust.transport.drivers.confluent
+
+.. automodule:: faust.transport.drivers.confluent
+    :members:
+    :undoc-members:

docs/reference/faust.transport.kafkaesq.rst

@@ -0,0 +1,11 @@
+=====================================================
+ ``faust.transport.kafkaesq``
+=====================================================
+
+.. contents::
+    :local:
+.. currentmodule:: faust.transport.kafkaesq
+
+.. automodule:: faust.transport.kafkaesq
+    :members:
+    :undoc-members:

docs/reference/index.rst

@@ -159,6 +159,8 @@ Transports
     faust.transport.producer
     faust.transport.drivers
     faust.transport.drivers.aiokafka
+    faust.transport.drivers.confluent
+    faust.transport.kafkaesq
     faust.transport.utils
 
 Assignor