feature(Streaming): Add Kafka live-stream wrapper and acceptance test

[szemyd]

Thin wrapper over confluent-kafka for the live metric streams, plus an acceptance test covering the external customer authentication we just shipped.

Written to be shareable: the wrapper is documented in the README with a runnable example, so it can go to a trial customer as-is rather than only serving CI.

Secrets to configure

In this repository's settings, before the acceptance job can do anything (it skips cleanly until then). Names only — values belong in the secret store:

Secret	Purpose
APERIODIC_KAFKA_BOOTSTRAP	Kafka bootstrap endpoint, host:port
APERIODIC_KAFKA_ACCOUNT_ID	Test account id — the SASL username
APERIODIC_KAFKA_API_KEY	That account's Aperiodic data API key — the SASL password

Use a dedicated throwaway test account, never a real customer's.

Optional repository variable (not a secret) — when set, the topic listing is checked exactly rather than loosely:

Variable	Purpose
APERIODIC_KAFKA_ENTITLED_DATASETS	Comma-separated dataset prefixes the test account's tier covers, e.g. ohlcv,l1_liquidity

No router-side secret is needed — see the companion PR on dream-faster/unravel-router.

The wrapper — src/aperiodic/streaming/

• Config is built from APERIODIC_KAFKA_* with no fallback defaults. A hard-coded fallback in a public repo publishes the value it falls back to. A missing variable raises StreamConfigError naming the variable, never its value.
• Enforces the <account-id>-<suffix> consumer group rule. Callers pass group_suffix= and the prefix is applied, so the most likely mistake — a GROUP_AUTHORIZATION_FAILED from a hand-written group id — cannot be made.
• Every network call is bounded.
• stream() holds one consumer open for continuous reading; consume() is the bounded convenience built on it. Repeated consume() calls would rejoin the group and rebalance each time, which matters for a customer following a topic.
• SASL_SSL / SCRAM-SHA-256, system CA store, verification explicitly on and no ssl.ca.location.

Redaction is the reason it exists

librdkafka embeds the broker address in nearly every error and log line it produces, and writes them to stderr from C, below the reach of Python-level capture. This is a public repo with world-readable CI logs.

Two layers, because one is not enough:

1. Literal replacement of the endpoint, account id and API key.
2. Structural replacement of anything address-shaped. This is the layer that matters — a cluster advertises brokers under hostnames the client never configured, so literal matching has nothing to match, and neither does GitHub's secret masking.

Also: librdkafka is routed through a redacting logger rather than stderr, __repr__ redacts, the config dict is never logged, and translated errors are re-raised with from None — from exc would reprint librdkafka's message, broker address and all, in the traceback.

Verified against the real thing, not just stubs: test_forced_connection_failure_leaks_nothing drives librdkafka into a genuine connection failure against a closed local port and asserts the endpoint appears in no surface — exception, traceback, stdout, stderr (via capfd, since capsys cannot see C-level writes) or log records. librdkafka does emit its FAIL records on that path, so the filter is doing real work.

The redaction rules are deliberately eager, and one test (test_useful_diagnostics_survive) holds the other side of that line so error messages stay useful. It caught a real over-match during development: ohlcv.binance-futures.m1 was being read as an FQDN and scrubbed to [redacted-host]-futures.m1.

The acceptance test — tests/test_kafka_acceptance.py

Asserts access control, not message flow:

1. Cluster metadata round trip — proves DNS + TLS + SASL end to end.
2. Topic listing is well-formed and contains only entitled dataset prefixes.
3. An unentitled dataset is denied with TOPIC_AUTHORIZATION_FAILED.
4. Wrong password is rejected with SASL_AUTHENTICATION_FAILED / _AUTHENTICATION.
5. A consumer group outside the account prefix is rejected — the wrapper refuses to build one, so this asserts the wrapper raises.

There is no blocking consume assertion. Staging metric topics are currently empty, so one would hang and read as an auth failure. The one read that does happen is short and bounded, and treats "no messages" as a pass — consume() distinguishes a quiet topic (returns []) from being refused or never connecting (raises), so an empty topic cannot mask a broken connection.

Skips, never fails, when the env vars are absent — that is what keeps fork PRs green.

No assertion embeds the endpoint, account id or API key: pytest prints the operands of a failing assertion, so those comparisons are reduced to a boolean first, and ClusterInfo carries a broker count rather than broker names for exactly this reason.

Workflow — .github/workflows/kafka-acceptance.yaml

Two jobs. unit-tests runs the redaction suite with no secrets and no network, and gates acceptance, which carries the secrets.

No pull_request trigger, so fork PRs never reach it; the acceptance job is additionally pinned to this repository so a fork pushing to its own main cannot run it. No set -x, no echo of any secret, and an add-mask step registers the values derived from secrets — the bare host without its port, and the account id — since GitHub masks only the literal secret.

Verification

• pytest tests/test_streaming.py — 50 passed
• pytest tests/test_kafka_acceptance.py — 7 skipped (no credentials locally, as intended)
• ruff check . — clean; full suite still collects at 143 tests
• Grepped the committed diff for host:port literals, broker-shaped hostnames, public IPs, UUIDs and key-shaped assignments: only invented fixtures on RFC 2606 reserved domains

confluent-kafka is a new [streaming] extra, kept out of [tests] on purpose — it is a compiled wheel and the existing matrix spans Python versions it does not always ship for.

---

Generated by Claude Code