Readd Confluent Kafka transport driver (rebased on master, working on released confluent-kafka)#679
Merged
Merged
Conversation
Resolve conflict in faust/transport/drivers/__init__.py by keeping the FactoryMapping-based transport registry (with the confluent driver registered), consistent with every other faust registry (stores, web cache backends, fixups) and preserving lazy driver loading so the optional confluent_kafka dependency is only imported when the confluent transport is actually selected.
The driver depended on Consumer.io_event_enable(), an eventfd-based notification API that only exists in the unmerged 2022 fork stephan-hof/confluent-kafka-python@features/io_event_enable and is absent from every released confluent-kafka. This made faust[ckafka] install an abandoned personal fork and the driver unusable on the official package. Remove the eventfd notification machinery from AsyncConsumer (eventfd, io_event_enable, add_reader/__eventfd_ready, the waiters set and the weakref finalizer) and reduce poll() to a thin wrapper over the blocking confluent_kafka.Consumer.poll(). This mechanism was already unused by the real fetch path -- ConfluentConsumerThread.getmany() drives the blocking Consumer.consume() inside the dedicated consumer thread via call_thread() -- so no functional behavior changes. Point requirements/extras/ckafka.txt at the official confluent-kafka>=2.0.0 release, and drop the now-unused os, struct and weakref imports plus a dead cast in Consumer.on_stop().
The previous tests/unit/transport/drivers/test_confluent.py (~1375 lines) was copied wholesale from the aiokafka test suite and asserted against methods the confluent driver never had (_fetch_records, _max_poll_records, transactional producer methods that raise NotImplementedError, ...), so 60 of 85 tests failed and several blocked in librdkafka. Replace it with a focused suite that exercises the driver's actual surface -- server_list, Consumer, AsyncConsumer, ConfluentConsumerThread, Producer, ProducerThread, ProducerProduceFuture and Transport -- with the underlying confluent_kafka Consumer/Producer mocked so no broker is needed. 56 tests, all passing, runs in <1s. Includes a regression guard that AsyncConsumer construction never calls the removed io_event_enable.
Consumer.create_topic() computed retention as int(want_seconds(retention) * 1000.0) unconditionally, so the default retention=None raised TypeError (None * 1000.0) whenever a topic was declared without an explicit retention. Guard the conversion the same way the aiokafka driver does (pass None through), and add a regression test.
confluent-kafka is an optional extra (faust[ckafka]) and is not part of
requirements/test.txt, so it is absent in the CI test matrix. The test
module imports the driver (which imports confluent_kafka) at import time,
which turned every matrix job into a collection error. Guard the module
with pytest.importorskip('confluent_kafka') so it skips cleanly when the
optional dependency is unavailable, and runs in full where it is.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #679 +/- ##
=======================================
Coverage 94.09% 94.10%
=======================================
Files 102 102
Lines 11108 11107 -1
Branches 1198 1198
=======================================
Hits 10452 10452
+ Misses 557 556 -1
Partials 99 99 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
confluent.py is only importable when the ckafka extra (faust[ckafka]) is installed, which the CI test environment does not do -- so its tests are skipped there and the file shows as 0% covered, dragging project/patch coverage down. Omit it from coverage like the memory driver already is; the driver's own tests still run wherever confluent-kafka is available.
The confluent readd had switched examples/word_count.py to 'confluent://localhost:9092'. Confluent is an optional, experimental transport (installed via faust[ckafka]); mainstream examples should use the default aiokafka transport. Revert the example broker to 'kafka://localhost:9092'.
wbarnha
marked this pull request as ready for review
July 17, 2026 02:43
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Reintroduces the
confluenttransport driver (faust.transport.drivers.confluent), superseding the stale 2022 draft #418. That PR had drifted years behindmasterand depended on an unreleased fork ofconfluent-kafka; this branch brings the work up to currentmasterand makes the driver run on the official package.What's here
faust/transport/drivers/__init__.py): restores theFactoryMapping-based transport registry withaiokafka,kafkaandconfluentregistered. This matches every other faust registry (stores,web/cache/backends,fixups) and keeps driver loading lazy, so the optionalconfluent_kafkadependency is only imported when theconfluent://transport is actually selected.confluent-kafka: the previous driver calledConsumer.io_event_enable(), an eventfd-based notification API that exists only in the unmerged forkstephan-hof/confluent-kafka-python@features/io_event_enableand is absent from every publishedconfluent-kafka. That machinery is removed fromAsyncConsumer. It was already unused by the real fetch path —ConfluentConsumerThread.getmany()drives the blockingConsumer.consume()on the dedicated consumer thread viacall_thread()— so there is no functional behavior change.poll()is now a thin wrapper over the blockingConsumer.poll().requirements/extras/ckafka.txtnow points at the officialconfluent-kafka>=2.0.0release instead of the git fork.create_topicretention fix:Consumer.create_topic()computedretentionasint(want_seconds(retention) * 1000.0)unconditionally, which crashed on the defaultretention=None. Now guarded to passNonethrough, matching the aiokafka driver (its own commit in this branch).tests/unit/transport/drivers/test_confluent.pywas copied wholesale from the aiokafka suite and asserted against methods the confluent driver never had (_fetch_records, transactional producer methods, …), so most of it failed. Replaced with a focused, broker-free suite (57 tests, all passing, <1s) coveringserver_list,Consumer,AsyncConsumer,ConfluentConsumerThread,Producer,ProducerThread,ProducerProduceFutureandTransport, with the underlyingconfluent_kafkaobjects mocked. Includes a regression guard thatAsyncConsumerconstruction never calls the removedio_event_enable.Status / notes
confluent-kafka2.15.0. It has not been exercised against a live broker end-to-end, and the producer's transactional/create_topicpaths remainXXXno-ops carried over from the original driver — worth a follow-up before this leaves draft.Supersedes #418.
Generated by Claude Code