Address review feedback on kafka adapter CI

Restore the mapper re-exports on csp.adapters.kafka that were dropped in favour of __all__.
__all__ only governs star imports, so `from csp.adapters.kafka import DateTimeType` had started
raising ImportError, breaking examples/03_using_adapters/kafka. Also fix that example's reference
to ProtoMessageMapper, which has never existed under that name.

Create test topics via the Kafka AdminClient instead of publishing a warm-up record. The old
helper wrote a record under key "foo", which test_metadata and test_push_mode both subscribe to.
Deterministic creation also lets the broker disable auto-creation, which un-skips test_invalid_topic.

Restore delivery coverage in test_basic and test_raw_pubsub. Tail-slicing pub against sub only
lines up when the startup gap equals the in-flight lag, and set membership cannot detect loss,
duplication or reordering. Both now align on the first record theduplication or reordering. Both now align on the first record theduplication os duplication or reordering. Both now a the broduplication or reordering. Both now align on the first record theduplication or reo_REBALANCE_DELAY_MS=0, bind only 9092
to loopback, and drop the JMX port and the no-op ACL authorizer.

Pin the new job's actions to the same SHAs used elsewhere in the workflow, and scope it with
read-only permissions and a timeout.

Signed-off-by: Tim Paine <3105306+timkpaine@users.noreply.github.com>

Author: timkpaine

.github/workflows/build.yml

@@ -717,6 +717,11 @@ jobs:
         - initialize
         - build
 
+      permissions:
+        contents: read
+
+      timeout-minutes: 30
+
       strategy:
         matrix:
           os:
@@ -730,25 +735,27 @@ jobs:
 
       steps:
         - name: Checkout
-          uses: actions/checkout@v4
+          uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
           with:
               submodules: recursive
+              persist-credentials: false
 
         - name: Set up Python ${{ matrix.python-version }}
           uses: ./.github/actions/setup-python
           with:
             version: '${{ matrix.python-version }}'
+            cibuildwheel: false
 
         - name: Install python dependencies
           run: make requirements
 
         - name: Install test dependencies
           shell: bash
-          run: sudo apt-get install graphviz
+          run: sudo apt-get update && sudo apt-get install -y graphviz
 
         # Download artifact
         - name: Download wheel
-          uses: actions/download-artifact@v4
+          uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
           with:
             name: csp-dist-${{ runner.os }}-${{ runner.arch }}-${{ matrix.python-version }}
 
@@ -757,16 +764,16 @@ jobs:
             python -m pip install -U *manylinux*.whl
             python -m pip install -U --no-deps *manylinux*.whl --target .
 
+        # Services declare healthchecks, so --wait blocks until the broker actually accepts connections
         - name: Spin up adapter service
-          run: make dockerup ADAPTER=${{ matrix.adapter }} DOCKERARGS="--wait --wait-timeout 30"
-
-        - name: Wait a few seconds after docker images have been spun up
-          run: sleep 30
+          run: make dockerup ADAPTER=${{ matrix.adapter }} DOCKERARGS="--wait --wait-timeout 180"
 
         # Run tests
         - name: Setup test flags
           shell: bash
-          run: echo "CSP_TEST_$( echo ${{ matrix.adapter }} | awk '{print toupper($0)}' )=1" >> $GITHUB_ENV
+          env:
+            ADAPTER: ${{ matrix.adapter }}
+          run: echo "CSP_TEST_${ADAPTER^^}=1" >> "$GITHUB_ENV"
 
         - name: Python Test Steps
           run: make test-py TEST_ARGS="-k ${{ matrix.adapter }}"

Makefile

@@ -117,7 +117,8 @@ tests: test
 
 .PHONY: dockerup dockerps dockerdown initpodmanmac
 ADAPTER := kafka
-DOCKER_COMPOSE := docker compose # or podman-compose
+# Prefer docker, fall back to podman-compose; override with DOCKER_COMPOSE=...
+DOCKER_COMPOSE := $(shell command -v docker >/dev/null 2>&1 && echo "docker compose" || echo "podman-compose")
 DOCKERARGS :=
 
 initpodmanmac:

ci/kafka/docker-compose.yml

@@ -1,40 +1,47 @@
 # https://github.com/conduktor/kafka-stack-docker-compose
 ---
-version: '2.1'
-
 services:
   zoo1:
-    image: confluentinc/cp-zookeeper:7.3.2
+    image: confluentinc/cp-zookeeper:7.5.3
     hostname: zoo1
     container_name: zoo1
     ports:
-      - "2181:2181"
+      - "127.0.0.1:2181:2181"
     environment:
       ZOOKEEPER_CLIENT_PORT: 2181
       ZOOKEEPER_SERVER_ID: 1
       ZOOKEEPER_SERVERS: zoo1:2888:3888
+    healthcheck:
+      test: ["CMD-SHELL", "echo ruok | nc -w 2 localhost 2181 | grep imok"]
+      interval: 5s
+      timeout: 5s
+      retries: 24
 
   kafka1:
-    image: confluentinc/cp-kafka:7.3.2
+    image: confluentinc/cp-kafka:7.5.3
     hostname: kafka1
     container_name: kafka1
     ports:
-      - "9092:9092"
-      - "29092:29092"
-      - "9999:9999"
+      - "127.0.0.1:9092:9092"
     environment:
-      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:19092,EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092,DOCKER://host.docker.internal:29092
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:19092,EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
       KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"
       KAFKA_BROKER_ID: 1
       KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
       KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
-      KAFKA_JMX_PORT: 9999
-      KAFKA_JMX_HOSTNAME: ${DOCKER_HOST_IP:-127.0.0.1}
-      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
-      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
+      # Consumer groups are created per-run, so skip the 3s default rebalance debounce
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      # Tests create their topics explicitly; test_invalid_topic depends on this being off
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
+    healthcheck:
+      test: ["CMD-SHELL", "kafka-broker-api-versions --bootstrap-server localhost:9092"]
+      interval: 5s
+      timeout: 10s
+      retries: 24
     depends_on:
-      - zoo1
+      zoo1:
+        condition: service_healthy

conda/dev-environment-unix.yml

@@ -17,7 +17,6 @@ dependencies:
  - flex
  - graphviz
  - gtest
- - httpx>=0.20,<1
  - libarrow<24
  - libboost>=1.80.0
  - libboost-headers>=1.80.0
@@ -41,6 +40,7 @@ dependencies:
  - pytest-sugar
  - python>=3.10,<3.15
  - python-build
+ - python-confluent-kafka
  - python-graphviz
  - python-rapidjson
  - pytz

conda/dev-environment-win.yml

@@ -17,7 +17,6 @@ dependencies:
 #  - flex  # not available on windows
  - graphviz
  - gtest
- - httpx>=0.20,<1
  - libarrow<24
  - libboost>=1.80.0
  - libboost-headers>=1.80.0
@@ -41,6 +40,7 @@ dependencies:
  - pytest-sugar
  - python>=3.10,<3.14
  - python-build
+ - python-confluent-kafka
  - python-graphviz
  - python-rapidjson
  - pytz

csp/adapters/kafka.py

@@ -8,11 +8,29 @@
 import csp
 from csp import ts
 from csp.adapters.status import Status
-from csp.adapters.utils import MsgMapper, hash_mutable
+from csp.adapters.utils import (
+    BytesMessageProtoMapper,
+    DateTimeType,
+    JSONTextMessageMapper,
+    MsgMapper,
+    RawBytesMessageMapper,
+    RawTextMessageMapper,
+    hash_mutable,
+)
 from csp.impl.wiring import ReplayMode, input_adapter_def, output_adapter_def, status_adapter_def
 from csp.lib import _kafkaadapterimpl
 
-__all__ = ("KafkaStatusMessageType", "KafkaStartOffset", "KafkaAdapterManager")
+__all__ = (
+    "KafkaAdapterManager",
+    "KafkaStartOffset",
+    "KafkaStatusMessageType",
+    # re-exported from csp.adapters.utils for backwards compatibility
+    "BytesMessageProtoMapper",
+    "DateTimeType",
+    "JSONTextMessageMapper",
+    "RawBytesMessageMapper",
+    "RawTextMessageMapper",
+)
 
 T = TypeVar("T")
 

csp/impl/wiring/adapters.py

@@ -1,6 +1,6 @@
 import inspect
 from datetime import timedelta
-from typing import List, TypeVar
+from typing import List
 
 from typing_extensions import override
 
@@ -14,7 +14,6 @@
 from csp.impl.wiring.signature import Signature
 
 _ = ReplayMode
-T = TypeVar("T")
 
 
 # Every AdapterDefMeta instance represents an input or output adapter *definition* type

csp/tests/adapters/conftest.py

@@ -1,3 +1,5 @@
+from uuid import uuid4
+
 import pytest
 
 from csp.adapters.kafka import KafkaAdapterManager
@@ -11,13 +13,10 @@ def kafkabroker():
 
 @pytest.fixture(scope="module", autouse=True)
 def kafkaadapterkwargs(kafkabroker):
-    return dict(broker=kafkabroker, group_id="group.id123", rd_kafka_conf_options={"allow.auto.create.topics": "true"})
+    # Unique group id so a rerun never inherits committed offsets from a previous run
+    return dict(broker=kafkabroker, group_id=f"csp.test.{uuid4()}")
 
 
 @pytest.fixture(scope="module", autouse=True)
-def kafkaadapter(kafkabroker):
-    group_id = "group.id123"
-    _kafkaadapter = KafkaAdapterManager(
-        broker=kafkabroker, group_id=group_id, rd_kafka_conf_options={"allow.auto.create.topics": "true"}
-    )
-    return _kafkaadapter
+def kafkaadapter(kafkaadapterkwargs):
+    return KafkaAdapterManager(**kafkaadapterkwargs)

csp/tests/adapters/kafka_utils.py

@@ -1,16 +1,16 @@
-from datetime import datetime, timedelta
+__all__ = ("create_topic",)
 
-import csp
-from csp.adapters.utils import DateTimeType, JSONTextMessageMapper
 
-__all__ = ("_precreate_topic",)
+def create_topic(broker, topic):
+    """Create `topic` and block until the broker acknowledges it.
 
+    Creation is done out of band rather than by publishing a warm-up message, so that no test data
+    lands on the topic and so tests do not depend on broker-side auto-creation (disabled in
+    ci/kafka/docker-compose.yml so test_invalid_topic can exercise the failure path).
+    """
+    # Imported lazily so collection does not require confluent-kafka when the kafka tests are skipped
+    from confluent_kafka.admin import AdminClient, NewTopic
 
-def _precreate_topic(adapter, topic):
-    """Since we test against confluent kafka, just use the kafka rest addon"""
-
-    def g():
-        msg_mapper = JSONTextMessageMapper(datetime_type=DateTimeType.UINT64_MICROS)
-        adapter.publish(msg_mapper, topic, "foo", csp.const("test"), field_map="a")
-
-    csp.run(g, starttime=datetime.utcnow(), endtime=timedelta(), realtime=True)
+    admin = AdminClient({"bootstrap.servers": broker})
+    for _, future in admin.create_topics([NewTopic(topic, num_partitions=1, replication_factor=1)]).items():
+        future.result()