Align confluent driver type annotations with the aiokafka driver

[wbarnha]

Description

faust/transport/drivers/confluent.py was the only transport driver still carrying implicit-Optional parameter defaults and fully unannotated functions — patterns absent from faust/transport/drivers/aiokafka.py sitting next to it. This PR brings the confluent driver's annotations up to the same standard, without reworking its logic.

What changed

• Explicit Optional[...] for every None default — config, retention, compacting, deleting, transactional_id, partition. PEP 484 prohibits implicit Optional, so these were also Liskov violations against the ConsumerT / ProducerT / ConsumerThread supertypes, which already declare the widened types. Affects Consumer.create_topic, ConfluentConsumerThread.create_topic, ConfluentConsumerThread.key_partition, Producer.create_topic, Producer.send, Producer.send_and_wait, and Transport._topic_config.
• AsyncConsumer is now annotated — __init__ plus close / subscribe / assign / poll, reusing the PartitionsRevokedCallback and PartitionsAssignedCallback aliases faust already defines in faust.types.transports.
• AsyncConsumer.assignment() returns List[TopicPartition], which is what confluent_kafka.Consumer.assignment() actually returns, rather than Set[TP]. ConfluentConsumerThread.assignment() already funnels the result through ensure_TPset().
• earliest_offsets / highwaters (and their helpers) return Mapping[TP, int], matching both the ConsumerThread base class and the aiokafka driver, instead of needlessly narrowing to MutableMapping.
• ProducerThread.produce takes optional key/value/partition — which is what the if partition is not None branch in its own body already assumes. on_delivery is deliberately left as a bare Callable; see below.
• ConfluentConsumerThread.key_partition reaches the consumer via _ensure_consumer(), like every other method in the class, so an unstarted thread raises ConsumerNotStarted rather than an AttributeError on None. This is the only behaviour change in the PR, and it mirrors AIOKafkaConsumerThread.key_partition.

Effect on mypy

Numbers depend on the mypy version, because 2.x reads inline types from py.typed packages (including confluent_kafka) that 1.x ignored. Both directions are an improvement:

	confluent.py	whole package
mypy 2.3.0 (current release)	53 → 17	692 → 656
mypy 1.19.1	38 → 0	552 → 514

A file-by-file diff of the full error list confirms no other module changes in either case.

The 17 remaining under 2.3.0 are all pre-existing defects, and several become visible only because of this PR: annotating AsyncConsumer.__init__ lets mypy resolve self.consumer to the real confluent_kafka.Consumer instead of Any, so call sites that were previously unchecked now get checked. What it surfaces is genuine — AsyncConsumer defines neither seek nor seek_to_beginning, yet _seek_wait and seek_to_beginning call both; and several sync callables are handed to call_thread, which expects awaitables.

On on_delivery

An earlier revision of this branch annotated it Callable[[Optional[BaseException], _Message], None], reasoning from ProducerProduceFuture.set_from_on_delivery's own signature. That is wrong: confluent's stubs declare Callable[[Optional[KafkaError], Message], None], and KafkaError does not derive from BaseException — KafkaException is the exception type, KafkaError is a plain status object. So set_from_on_delivery's self.set_exception(err) is handed a non-exception and raises TypeError instead of failing the future with the delivery error. The # XXX Not sure what err is here comment above it turns out to have been well founded. Fixing that is a behaviour change and belongs in its own patch, so the parameter is left as a bare Callable with a comment recording why.

Other latent bugs found, not fixed here

• Producer.send calls ProducerThread.produce(topic, value, key, partition) against a (topic, key, value, partition) signature — key and value are swapped.
• Producer.key_partition calls self._producer_thread.producer.list_topics(...), but ProducerThread.producer is the faust Producer, which has no such method.

Verification

• pytest tests/unit/transport/drivers/test_confluent.py — 57 passed (the same suite the dedicated confluent CI leg runs).
• isort --check, black --check, and flake8 clean on the changed file.

No issue to link; this is a type-annotation cleanup. Follow-up #757 wires mypy into CI behind a ratchet so this class of regression is caught automatically.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 96.06%. Comparing base (e600bb5) to head (ca2dc4e).
⚠️ Report is 1 commits behind head on master.

Additional details and impacted files

@@           Coverage Diff           @@
##           master     #756   +/-   ##
=======================================
  Coverage   96.06%   96.06%           
=======================================
  Files         103      103           
  Lines       11072    11072           
  Branches     1191     1191           
=======================================
  Hits        10636    10636           
  Misses        345      345           
  Partials       91       91           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.