ci: add live-broker integration test harness (Kafka in CI)

.github/workflows/python-package.yml

@@ -97,6 +97,75 @@ jobs:
         uses: codecov/codecov-action@v5
         with:
           token: ${{ secrets.CODECOV_TOKEN }}
+  test-integration:
+    name: 'Integration (Kafka ${{ matrix.kafka-version }})'
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    # Advisory for now: broker tests can be flaky while we stabilise them, so
+    # a red here must not block the required checks / the merge queue.  This
+    # job is intentionally NOT in the `check` job's `needs`.
+    continue-on-error: true
+    strategy:
+      fail-fast: false
+      matrix:
+        kafka-version: ['3.8.1']
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+          cache: pip
+          cache-dependency-path: requirements/test.txt
+      # Run Kafka as a plain container (not a GH `services:` container) so its
+      # very chatty broker log stays inside the container -- retrievable on
+      # demand via `docker logs` -- and the job log shows the pytest output.
+      - name: Start Kafka (KRaft, single node)
+        run: |
+          docker run -d --name kafka -p 9092:9092 \
+            -e KAFKA_NODE_ID=1 \
+            -e KAFKA_PROCESS_ROLES=broker,controller \
+            -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 \
+            -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
+            -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
+            -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
+            -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
+            -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
+            -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
+            -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
+            -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
+            -e KAFKA_AUTO_CREATE_TOPICS_ENABLE=true \
+            apache/kafka:${{ matrix.kafka-version }}
+      - name: Install dependencies
+        run: |
+          pip install -r requirements/test.txt
+          pip install -r requirements/extras/ckafka.txt
+          pip install .
+      - name: Wait for Kafka to be ready
+        run: |
+          python - <<'PY'
+          import socket, time, sys
+          deadline = time.monotonic() + 120
+          while time.monotonic() < deadline:
+              try:
+                  with socket.create_connection(("localhost", 9092), 2):
+                      print("Kafka port is open")
+                      break
+              except OSError:
+                  time.sleep(2)
+          else:
+              sys.exit("Kafka did not become reachable in time")
+          PY
+      - name: Run live-broker integration tests
+        env:
+          FAUST_TEST_BROKER: 'kafka://localhost:9092'
+        run: pytest tests/integration/broker -v -ra --tb=short --no-cov
+      # Only the tail of the broker log on failure, so it augments rather than
+      # buries the pytest output in the job log.
+      - name: Kafka broker logs (on failure)
+        if: failure()
+        run: docker logs --tail 40 kafka
   check:  # This job does nothing and is only used for the branch protection
     name: ✅ Ensure the required checks passing
     if: always()

tests/integration/broker/conftest.py

@@ -0,0 +1,80 @@
+"""Fixtures for live-broker integration tests.
+
+These tests need a real Kafka broker.  Point them at one with the
+``FAUST_TEST_BROKER`` environment variable (default
+``kafka://localhost:9092``).  When no broker is reachable every test in this
+package is skipped, so the suite stays green on developer machines and in the
+normal (broker-less) CI jobs -- the dedicated integration job in CI starts a
+Kafka service and runs them for real.
+"""
+
+import os
+import socket
+from uuid import uuid4
+
+import pytest
+
+DEFAULT_BROKER = "kafka://localhost:9092"
+
+
+def _broker_from_env() -> str:
+    return os.environ.get("FAUST_TEST_BROKER", DEFAULT_BROKER)
+
+
+def _bootstrap_from_url(url: str) -> str:
+    # kafka://host:port  ->  host:port  (aiokafka bootstrap_servers form)
+    hostport = url.split("://", 1)[-1]
+    hostport = hostport.split("/", 1)[0]
+    return hostport or "localhost:9092"
+
+
+def _broker_reachable(url: str) -> bool:
+    host, _, port = _bootstrap_from_url(url).partition(":")
+    try:
+        with socket.create_connection((host or "localhost", int(port or 9092)), 2):
+            return True
+    except OSError:
+        return False
+
+
+@pytest.fixture(scope="session")
+def broker_url() -> str:
+    """Return the broker URL, skipping the test if it is not reachable."""
+    url = _broker_from_env()
+    if not _broker_reachable(url):
+        pytest.skip(
+            f"No Kafka broker reachable at {url} - "
+            "set FAUST_TEST_BROKER to run live-broker integration tests"
+        )
+    return url
+
+
+@pytest.fixture(scope="session")
+def kafka_bootstrap(broker_url: str) -> str:
+    """Return ``host:port`` for aiokafka's ``bootstrap_servers``."""
+    return _bootstrap_from_url(broker_url)
+
+
+@pytest.fixture
+def unique_topic() -> str:
+    """A fresh topic name so tests don't read each other's messages."""
+    return f"faust-it-{uuid4().hex}"
+
+
+@pytest.fixture
+def unique_group() -> str:
+    """A fresh consumer-group / app id per test."""
+    return f"faust-it-{uuid4().hex}"
+
+
+@pytest.fixture(autouse=True)
+def threads_not_lingering():
+    """Override the global no-lingering-threads guard for broker tests.
+
+    Talking to a real broker makes asyncio resolve ``localhost`` via
+    ``getaddrinfo`` in its default ``ThreadPoolExecutor``; that resolver
+    thread (``asyncio_0``) outlives a single test and is not a leak we can
+    act on.  The strict global guard in ``tests/conftest.py`` would fail an
+    otherwise-passing round-trip, so shadow it here with a no-op.
+    """
+    yield

tests/integration/broker/test_smoke.py

@@ -0,0 +1,91 @@
+"""Smoke tests proving the CI Kafka broker is up and reachable.
+
+These are the foundation for broker-dependent regression tests: a raw
+:pypi:`aiokafka` round-trip and a raw :pypi:`confluent-kafka` round-trip,
+one per transport-client library faust ships drivers for.  They stay green
+on developer machines and the broker-less CI jobs (the ``broker_url`` fixture
+skips when no Kafka is reachable) and run for real in the dedicated
+integration job.
+
+A full faust ``App`` end-to-end test (start an app, drive an agent, shut it
+down) is intentionally left to a follow-up: embedding the app lifecycle in
+pytest needs its own hardening and shouldn't gate this foundation.
+
+``confluent-kafka`` is an optional dependency; its test skips when it is not
+installed.
+"""
+
+import asyncio
+
+import pytest
+from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
+
+# The suite escalates ResourceWarning to an error (see pyproject.toml), but
+# aiokafka / confluent-kafka legitimately deal with sockets that can emit
+# ResourceWarning during teardown of a live-broker test.  Relax that here so a
+# leaked-socket warning doesn't mask the actual assertion.
+pytestmark = [
+    pytest.mark.asyncio,
+    pytest.mark.filterwarnings("ignore::ResourceWarning"),
+]
+
+# Enough for a cold broker + first fetch in CI, short enough that a genuine
+# hang fails fast instead of sitting at the timeout.
+RECV_TIMEOUT = 20.0
+
+
+async def test_aiokafka_roundtrip(kafka_bootstrap, unique_topic):
+    """Produce and consume one message straight through aiokafka."""
+    producer = AIOKafkaProducer(bootstrap_servers=kafka_bootstrap)
+    await producer.start()
+    try:
+        await producer.send_and_wait(unique_topic, b"ping")
+    finally:
+        await producer.stop()
+
+    consumer = AIOKafkaConsumer(
+        unique_topic,
+        bootstrap_servers=kafka_bootstrap,
+        auto_offset_reset="earliest",
+        enable_auto_commit=False,
+        group_id=None,
+    )
+    await consumer.start()
+    try:
+        msg = await asyncio.wait_for(consumer.getone(), timeout=RECV_TIMEOUT)
+    finally:
+        await consumer.stop()
+    assert msg.value == b"ping"
+
+
+async def test_confluent_roundtrip(kafka_bootstrap, unique_topic, unique_group):
+    """Produce and consume one message straight through confluent-kafka."""
+    pytest.importorskip("confluent_kafka")
+    from confluent_kafka import Consumer, Producer
+
+    producer = Producer({"bootstrap.servers": kafka_bootstrap})
+    producer.produce(unique_topic, b"ping")
+    producer.flush(RECV_TIMEOUT)
+
+    consumer = Consumer(
+        {
+            "bootstrap.servers": kafka_bootstrap,
+            "group.id": unique_group,
+            "auto.offset.reset": "earliest",
+            "enable.auto.commit": False,
+        }
+    )
+    consumer.subscribe([unique_topic])
+    loop = asyncio.get_event_loop()
+    deadline = loop.time() + RECV_TIMEOUT
+    value = None
+    try:
+        while loop.time() < deadline:
+            msg = consumer.poll(1.0)
+            if msg is None or msg.error():
+                continue
+            value = msg.value()
+            break
+    finally:
+        consumer.close()
+    assert value == b"ping"