Fix the remaining latent bugs found by the type checker

[wbarnha]

Six bugs across five modules, all surfaced by the type checker in #758 and left marked XXX there because fixing them changes runtime behaviour. One commit per bug, each with a regression test. Two follow-up commits fix problems an adversarial review found in the fixes themselves — see Review follow-ups below.

#758 has since been squash-merged, so this branch has been rebased onto master and now contains only the six fixes. Independent of #759, which fixes the two aiokafka crashes — they touch disjoint files and can merge in either order.

What was broken

Module	Bug
auth.py	SSLCredentials() raised TypeError — could not build a context at all
__init__.py	faust.version_info.major was 'v'; every field was wrong
tables/base.py	on_window_close never received the aggregated window data
tables/recovery.py	changelog events applied to an unrelated table; aborted-tx fixup silently skipped
cli/base.py	require_app = False commands crashed — faust completion unusable without -A
transport/drivers/confluent.py	key_partition dead on arrival, and partitioning by a per-process hash underneath; livelock detector raised every tick

SSLCredentials could not be constructed

purpose defaulted to None and went straight into ssl.create_default_context(purpose=...), which starts if not isinstance(purpose, _ASN1Object): raise TypeError(purpose). Any call that did not pass an explicit context raised — which defeats the cafile/capath/cadata parameters entirely. Now defaults to ssl.Purpose.SERVER_AUTH, the same default create_default_context() itself uses and the right one for a client verifying a broker; the tests assert the resulting context has check_hostname=True and verify_mode=CERT_REQUIRED.

faust.version_info held strings in its int fields

The regex groups (prefix, version, suffix) were splatted positionally into VersionInfo(major, minor, micro, ...), giving VersionInfo(major=None, minor='0.11.5', micro=''). Now parsed properly, with any non-numeric tail (dev1+g1234, rc1) going to releaselevel. An unparsable version degrades to VersionInfo(0, 0, 0, ...) instead of raising, so import faust can no longer fail on it.

This changes a public value. Code reading faust.version_info.minor as the version string must use faust.__version__, which is unchanged. Nothing in the repo consumes it and no docs reference it.

on_window_close never got its window data

_del_old_keys read _partition_timestamp_keys with the whole (start, end) tuple as the second key element; the map is keyed on range_end alone, written that way by _maybe_set_key_ttl and read that way by _maybe_del_key_ttl. The lookup could never hit, so triggered_windows was always [None, ...] and on_window_close only ever saw the raw per-key value.

User-visible: applications with an on_window_close handler that hold lists of events per window will start receiving the aggregated data the API always promised. Every other value shape is delivered exactly as before — see the first review follow-up.

Recovery applied events to whichever table came last

_slurp_changelogs binds table/offsets/bufsize per TP. The else: branch for an untracked TP only logged a warning and fell through, so the event was applied using the previous iteration's bindings — or raised UnboundLocalError if it was the first event. Now skipped. Worth noting a bare continue would be wrong: _maybe_signal_recovery_end() and the standby bookkeeping at the bottom of the loop must still run every iteration, so only the application block is skipped.

Separately, detect_aborted_tx compared await consumer.position(tp) >= highwater unguarded. position is Optional[int] and does return None, raising TypeError — swallowed by the caller's except Exception, which then skipped the aborted-transaction fixup for every remaining partition.

require_app = False was unusable

_app_from_str returns None for such a command invoked without -A, then AppCommand.__init__ did key_serializer or self.app.conf.key_serializer unconditionally. self.app is now a property over an Optional[AppT], with on_stop and blocking_timeout no longer assuming an app. Behaviour with an app present is unchanged, and require_app = True still raises the same UsageError.

confluent driver

key_partition reached for list_topics on ProducerThread.producer — the Faust producer — rather than ._producer, the confluent handle. Faust producers have no list_topics, so the method raised AttributeError and was dead on arrival. Read the confluent producer instead. That unblocked the partition arithmetic underneath it, which had its own bugs — see the second review follow-up.

And Consumer.verify_event_path delegated to a method no thread class defines, so the livelock detector raised every tick; it now has the same documented no-op the base Consumer.verify_event_path already is. A real livelock implementation is separate work.

Review follow-ups

An adversarial review of this branch found two problems in the fixes above. Both are confirmed and fixed here.

Window cleanup crashed on values that are not lists

Correcting the _partition_timestamp_keys lookup made the aggregation block reachable for the first time, and it assumed every window value is a list:

window_data.setdefault(processed_window[0], []).extend(
    self.data.get(processed_window, [])
)

For a counting table — app.Table("counts", default=int).hopping(...), the shape the docs use — the stored value is an int, and list.extend(2) raises TypeError: 'int' object is not iterable. That kills the _clean_data background task, and the crash is not confined to applications that define an on_window_close handler: window_data is built before any callback dispatch.

A string is worse than a crash: it is iterable, so "BOO" and "MOO" across two overlapping windows silently aggregate to ['B', 'O', 'O', 'M', 'O', 'O'].

Only list values are aggregated now. Anything else is left out of window_data and delivered as the raw per-key value — exactly what handlers received for as long as the lookup was broken, so no scalar or string table changes behaviour.

confluent key_partition was not deterministic

With the AttributeError out of the way, the arithmetic underneath ran for the first time:

key_bytes = str(key).encode("utf-8")
partition = abs(hash(key_bytes)) % partition_count

Two bugs in three lines.

hash() is salted per process for bytes, so the same key resolved to a different partition in every worker and after every restart: abs(hash(str(b"key").encode())) % 2 gives 1, 0, 0 under PYTHONHASHSEED 0, 1 and 12345. key_partition is what PartitionAssignor.key_store uses to route a request for a key to the worker owning that key's partition, and what Channel.send uses under eager_partitioning; an unstable answer sends both to the wrong place.

And str(key) on a bytes key renders its repr, so b"key" was hashed as the eight bytes of "b'key'" rather than the three the broker sees.

Both are replaced by partition_for_key, crc32(key) % partition_count. That is librdkafka's default consistent_random partitioner, which this driver never overrides, so a computed partition now matches where a keyed record produced through it actually lands. Deliberately not the Java client's murmur2 — that is what aiokafka's DefaultPartitioner uses, and each driver has to mirror the library that will do its producing.

ConfluentConsumerThread.key_partition carried the identical arithmetic and is fixed with it; leaving it would have the driver's consumer and producer disagree about where a key belongs.

The tests asserted only 0 <= partition < n, which any hash function passes, salted or not. They now assert the fixed partition crc32 picks, plus vectors for the helper and a cross-process check under three different PYTHONHASHSEEDs.

Verification

scripts/check clean. mypy -p faust clean without the ckafka extra installed, which is the configuration the lint job runs in; installing it surfaces 12 pre-existing errors in confluent.py's seek/getmany paths, identical before and after this branch and untouched by it.

Measured locally with the ckafka extra installed, so the confluent driver's tests actually execute rather than skipping:

• tests/unit/transport/drivers/test_confluent.py: 68 passed
• everything else (tests/unit functional integration meticulous regression): 2258 passed, 7 skipped (3 aiokafka-specific — two documented source bugs and one removed api_version — plus 4 needing a live Kafka or Redis)

Every new test was checked against the pre-fix code by overlaying the test files onto a pristine tree built from git archive. All fail there except the one noted below.

The three window-cleanup tests were checked the same way against this branch's own pre-follow-up state: all three fail there, on TypeError: 'int' object is not iterable and on the shredded string. The confluent partitioning tests cannot be run that way — they import a helper that did not exist before the fix — so the nondeterminism is evidenced directly by the PYTHONHASHSEED figures above, and the new assertions are fixed values that the old salted arithmetic cannot reproduce.

One caveat: tests/unit/cli/test_base.py::test_init__serializers_without_app passes against pre-fix code too — both serializers it passes are truthy, so the old code short-circuits before dereferencing the None app. It documents the case but is not a guard; test_init__no_app_when_not_required is the one that actually fails pre-fix.

An earlier revision of this description said the confluent fixes were unverified by execution because CI does not install the extra. That is no longer true: master grew a dedicated confluent CI leg (Python 3.10–3.14) that installs requirements/extras/ckafka.txt and runs tests/unit/transport/drivers/test_confluent.py, and this branch picked it up in the merge from master.

No existing test was weakened or deleted. The only removed test lines are an unused import, three mock_ranges calls replaced with explicit (start, end) tuples (the mock returned bare floats, which is not what _window_ranges yields), and two mock-setup lines that pointed at the very attribute the confluent fix corrects.

🤖 Generated with Claude Code

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 96.10%. Comparing base (e9f102f) to head (232aa0d).

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #760      +/-   ##
==========================================
+ Coverage   96.05%   96.10%   +0.04%     
==========================================
  Files         104      104              
  Lines       11087    11116      +29     
  Branches     1189     1198       +9     
==========================================
+ Hits        10650    10683      +33     
+ Misses        345      343       -2     
+ Partials       92       90       -2     

☔ 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.