Skip to content

Commit db4ec50

Browse files
tcdentclaude
andauthored
0.2 architecture: class-based backends, partitioned queues, activity handlers
* Add Kafka stream backend architecture with split KV/stream protocols Introduces a layered backend architecture to support Kafka alongside Redis: - KVBackend protocol: key-value, counters, sorted sets, locks, pub/sub - StreamBackend protocol: produce/consume, topic management, compacted topics - Redis KV backend: extracted from redis_backend.py, implements KVBackend - Kafka stream backend: connection mgmt, produce/consume via aiokafka - Operations layer (ops.py): bridges agentexec modules to either backend, with lock no-ops when stream backend handles partition-based isolation - Config additions: kv_backend, stream_backend, kafka_* settings - Full backward compatibility: legacy state_backend path still works https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Simplify to single-backend architecture with ops layer wired to all callers Replaces the dual KV+stream model with a single backend choice: AGENTEXEC_STATE_BACKEND=agentexec.state.redis_backend (default) AGENTEXEC_STATE_BACKEND=agentexec.state.kafka_backend Key changes: - Unified StateBackend protocol with semantic ops (queue_push/queue_pop instead of rpush/lpush/brpop) - ops.py: thin delegation layer, no dual-mode branching - All callers (queue.py, schedule.py, tracker.py, worker/pool.py, worker/event.py, worker/logging.py, core/results.py) now go through ops instead of touching state.backend directly - kafka_backend.py: full implementation with compacted topics for KV, in-memory caches for sorted sets/counters, no-op locks - redis_backend.py: adds queue_push/queue_pop wrapping rpush/lpush/brpop - Removed dual-mode files: kv_backend.py, stream_backend.py, redis_kv_backend.py, kafka_stream_backend.py - Config simplified: single state_backend, no kv_backend/stream_backend state.backend still exported for backward compat with existing tests. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Add queue commit/nack semantics and retry support for Kafka resilience Key changes: - queue_commit(): acknowledges successful task processing (commits offset in Kafka, no-op in Redis) - queue_nack(): signals task should be retried (skips offset commit in Kafka, no-op in Redis). Task stays in its original partition position, preserving ordering. - Worker loop: commits on success, nacks on failure with retry tracking. After max_task_retries exhausted, commits to move past the message. - Task.retry_count field tracks attempt number - AGENTEXEC_MAX_TASK_RETRIES config (default 3) - task.py migrated from state.aset_result to ops.aset_result Kafka partition assignment acts as an implicit "in progress" marker — only the assigned consumer can read from its partitions, so no other worker can steal an uncommitted task. Redelivery only happens on consumer crash (heartbeat timeout) or explicit rebalance. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Concurrent task execution per worker and Kafka consumer heartbeat Restructures the worker loop to support concurrent task processing: - Worker._run() now spawns tasks as asyncio coroutines instead of awaiting them inline - asyncio.Semaphore caps concurrency at tasks_per_worker (default 1, backward compatible) - Poll loop stays active while tasks run, keeping Kafka consumer heartbeats alive for long-running AI agent tasks - In-flight tasks are awaited on shutdown for graceful completion New config: - AGENTEXEC_TASKS_PER_WORKER: max concurrent tasks per worker process Total concurrency = num_workers * tasks_per_worker This solves the Kafka partition-per-consumer constraint: instead of needing one process per partition, a single worker can own multiple partitions and process their tasks concurrently. Ideal for I/O-bound AI workloads where tasks spend most time waiting for LLM responses. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Revert "Concurrent task execution per worker and Kafka consumer heartbeat" This reverts commit a5a6584. * Add worker_id to Kafka client IDs for observability Each MP worker process now calls ops.configure(worker_id=...) on startup, which the Kafka backend uses to build unique client_id strings (e.g. agentexec-worker-0, agentexec-producer-1). This lets broker logs and monitoring tools distinguish between consumers in the same group. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Route activity system through ops layer with Kafka activity topic Activity lifecycle (create, update, list, detail) now goes through the ops layer like all other state operations, making it backend-agnostic. Kafka backend: activity records are produced to a compacted topic (agentexec.activity) keyed by agent_id. Each update appends to the log history and re-produces the full record. Pre-compaction, all intermediate states are visible; post-compaction, only the final state survives. In-memory cache serves queries. Redis backend: activity functions wrap the existing SQLAlchemy/Postgres logic with lazy imports to avoid circular dependencies. tracker.py: rewritten to delegate to ops instead of using SQLAlchemy directly. Session parameter kept for backward compatibility but ignored (backends manage their own sessions). https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Split backends into packages with domain-specific modules Each backend (redis_backend, kafka_backend) is now a package with: - connection.py: client/producer management and lifecycle - state.py: KV, counters, locks, pub/sub, sorted sets, serialization - queue.py: task queue push/pop/commit/nack - activity.py: task lifecycle tracking New protocols.py defines StateProtocol, QueueProtocol, and ActivityProtocol as separate domain contracts. backend.py validates that a backend implements all three. Import paths unchanged — agentexec.state.redis_backend and agentexec.state.kafka_backend still work via package __init__.py re-exports. ops.py and config remain untouched. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Go full async and rename backend methods to descriptive names Drop sync/async duality — all I/O methods are now async (no more `a` prefix). Rename Redis-ism method names to descriptive ones: get/set/delete → store_get/store_set/store_delete, incr/decr → counter_incr/counter_decr, zadd/zrangebyscore/zrem → index_add/ index_range/index_remove, publish/subscribe → log_publish/ log_subscribe. Pool.start() and Pool.shutdown() are now async, with schedule registration deferred to start(). All callers, protocols, and tests updated. 255 tests pass. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Add Kafka integration tests and CI workflow - CI workflow with two jobs: unit tests (fakeredis) and Kafka integration tests (real broker via bitnami/kafka:3.9 KRaft mode) - Integration tests cover: KV store, counters, sorted index, serialization, queue push/pop/commit, activity lifecycle, log pub/sub, and connection management - Add `kafka` optional dependency group (aiokafka>=0.11.0) - Tests skip gracefully when Kafka not available https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Fix CI: use correct Kafka image tag, override addopts, fix readiness check - bitnami/kafka:3.9 → 3.7 (3.9 doesn't exist) - Add -o "addopts=" to both pytest commands to avoid --ty/--cov conflicts - Switch Kafka readiness check from docker exec to nc -z https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Switch to apache/kafka:3.9.2 for CI Kafka service bitnami/kafka image failed to pull. apache/kafka is the official Apache Kafka Docker image with KRaft mode built in. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * CI: disable fail-fast, add verbose test output for debugging https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Fix Kafka consumer hangs: per-topic group IDs, retry loop, faster heartbeat - Use per-topic consumer group IDs to avoid cross-topic rebalancing - Add retry loop in queue_pop for partition assignment delays - Configure faster heartbeat (1s) and session timeout (10s) - Increase test queue_pop timeout to 10s for CI reliability https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Use manual partition assignment instead of consumer groups Consumer group protocol causes hangs during group-join/rebalance in CI. Manual partition assignment + explicit offset tracking eliminates group coordination overhead entirely. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Fix force_metadata_update — use partition discovery retry loop force_metadata_update doesn't exist on AIOKafkaConsumer in aiokafka 0.13.0. Replace with a retry loop that polls partitions_for_topic until metadata is available. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Remove consumer group_id to avoid group coordinator hangs group_id triggers GroupCoordinator even with manual partition assignment, causing hangs in CI. Remove it entirely — offset tracking is implicit via consumer position after getmany(). https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * CI: capture Kafka test output in job summary on failure https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * CI: post Kafka test output as PR comment on failure https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Use subscribe() with per-topic group IDs, set rebalance delay to 0 Manual partition assignment without group_id fails because metadata isn't fetched for unsubscribed topics. Switch back to subscribe() with per-topic group IDs. Also set group.initial.rebalance.delay.ms=0 on the CI broker for instant group joins. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Use admin metadata for partition discovery in manual assignment Consumer group protocol hangs reliably in CI. Use manual partition assignment with admin client describe_topics for reliable partition discovery instead of consumer metadata which requires subscription. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * CI: use curl for PR comment instead of github-script https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * CI: upload test output as artifact instead of PR comment https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Add debug prints to queue_pop and test_push_and_pop https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Update uv.lock after adding kafka extra dependency https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * CI: emit test output as warning annotations for API access https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Better debug output: print consumer state on timeout, filter annotations https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * CI: filter annotations to only show failures and debug output https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * CI: re-trigger after transient Docker pull failure https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * CI: retry after transient failures https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * CI: use apache/kafka:latest to avoid Docker pull issues with pinned tag https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * CI: switch to confluentinc/cp-kafka:7.7.1 for reliable Docker pulls apache/kafka image has persistent pull failures from GitHub Actions. Confluent Platform image is more widely available. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * CI: use docker run instead of service containers for Kafka Service containers use a separate Docker pull mechanism that's failing with rate limits. docker run in a step has better retry behavior and runs in parallel with dependency installation. https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Add docker-compose.kafka.yml, clean up debug prints, use docker run in CI - Add docker-compose.kafka.yml with recommended apache/kafka:3.9.0 setup - Remove debug print statements from queue.py and tests - CI uses docker run instead of service containers (more reliable pulls) - Update test docstring to reference docker-compose file https://claude.ai/code/session_015DuCUpx8r1TnLZo9dDUn4j * Fix queue_pop message buffer and produce() key type handling - Buffer messages from getmany() so multiple messages per batch aren't lost — getmany returns all available messages across partitions, but queue_pop should return one at a time - Accept bytes keys in produce() (not just str) All 27 Kafka integration tests now pass locally. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Kafka consumer groups, full async, producer-side topic creation Major refactor aligning the Kafka backend with idiomatic patterns: - Queue uses consumer groups for reliable fan-out across workers - All I/O is async — removed sync log_publish and produce_sync - Topic creation moved to produce side (ensure_topic in push paths) - Removed queue_commit/queue_nack — commit happens on pop, retries via explicit requeue with incremented retry_count - Proper typing throughout — real aiokafka types, UUID for agent_id - Stateless worker identity from hostname+pid, no cached globals - Simplified worker loop: early returns, exception-based retry - Dequeue hydrates Task directly (moved from worker to queue module) - docker-compose.kafka.yml stripped to pure Kafka bootstrap - Compacted topics with configurable retention (default: forever) - All 299 tests passing (272 unit + 27 Kafka integration) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Class-based backend architecture, eliminate ops passthrough layer - New base.py with ABCs: BaseBackend, BaseStateBackend, BaseQueueBackend, BaseActivityBackend. Shared serialize/deserialize in BaseBackend. - KafkaBackend and RedisBackend classes with namespaced sub-backends: backend.state, backend.queue, backend.activity - Public `backend` reference in state/__init__.py — callers import and use directly, no get_backend() indirection - Key constants (KEY_RESULT, KEY_LOCK, etc.) stay in state/__init__.py - Domain modules own their key formatting (schedule, event, results) - All ops.py passthrough functions eliminated - Connection state moved from module globals to instance attributes - count_active/get_pending_ids fixed to check last log status only - Test fixtures simplified: inject fake client via backend._client - 295 tests passing (268 unit + 27 Kafka integration) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Flatten backend modules, remove dead code, clean up noise - Flatten kafka_backend/ and redis_backend/ dirs to single files: state/kafka.py and state/redis.py - Backend class renamed to just Backend (module path is the qualifier) - Remove backend registry — _create_backend imports any module path with a Backend class, enabling custom backends - Config value simplified: agentexec.state.redis, agentexec.state.kafka - Delete dead files: ops.py, protocols.py, backend.py, and all old module-level state/queue/activity/connection files - Remove section separator comments and trivial file docstrings - Net -2093 lines deleted Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Kafka headers, stateless activity backend, schedule backend, pool supervision Major Kafka backend improvements: - ax_ prefixed headers on all produces (activity, queue, schedule) for metadata filtering without body deserialization - Activity backend reads directly from Kafka (no in-memory cache) with backwards-scan for single record lookup and offset-based pagination - Schedule backend with dedicated compacted topic (no more sorted set simulation) — persistent consumer with seek-to-beginning replay - Pool._supervise split into _process_log_stream and _process_scheduled_tasks with asyncio.gather - Pool.start() is now the foreground entry point, run() wraps it - Tick logic inlined in pool, removed from schedule.py - Schedule poll interval configurable (default 10s, was 100ms) - Log channel internalized in backends (no more CHANNEL_LOGS constant) - Status enum extracted to activity/status.py (no SQLAlchemy dependency) - Deprecation warnings on activity tracker session parameter - Docker compose updated with kafka-ui for development Skipped 3 Kafka integration tests (aggregate queries on shared topic) 267 unit + 24 kafka = 291 passing, 3 skipped Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Extract activity from backends into producer/consumer pattern Activity is no longer a backend concern. Workers produce events via generic pubsub, the pool's consumer writes to Postgres. Queries always hit Postgres regardless of backend. - activity/producer.py — event emitter called by workers - activity/consumer.py — pool-side Postgres writer - activity/__init__.py — query functions (list, detail, count_active) - Removed BaseActivityBackend and all backend activity implementations - Generalized log_publish/log_subscribe to publish/subscribe with channel parameter — reusable for logs, activity, and future streams - Pool.start() now runs three concurrent tasks: log stream, scheduled tasks, and activity stream - Removed Kafka activity_topic (no longer needed) - Removed Redis activity backend (Postgres is always the activity store) - Net -304 lines Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Typed worker messages, Task as pure data, multiprocessing IPC Major separation of concerns between Task, TaskDefinition, and Pool: - Task is pure data: task_name, context (Mapping), agent_id, retry_count No more _definition binding, execute(), or get_lock_key() on Task - TaskDefinition owns behavior: execute(task), get_lock_key(context), hydrate_context(). Looked up by task_name in the worker registry. - Worker → Pool communication via typed Message subclasses over multiprocessing.Queue: TaskCompleted, TaskFailed, LockContention, LogEntry. No more Redis pubsub for logs. - Pool._process_worker_events dispatches with match/case on message type - Removed _process_log_stream (logs flow through the same queue) - QueueLogHandler replaces StateLogHandler (writes to mp.Queue not pubsub) - Generalized log_publish/log_subscribe to publish/subscribe with channel - Lock key formatting and TTL moved into backend.state.acquire_lock - dequeue() no longer needs the task registry - Removed requeue() — pool handles requeueing via _process_worker_events 264 passed, 0 failed Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Partitioned Redis queues with scan-based fair dequeue Redis queue backend now partitions tasks by lock key: - Default queue: {queue_prefix} (lock-free, concurrent) - Partition queues: {queue_prefix}:{lock_key} (serialized by lock) - Locks: {queue_prefix}:{lock_key}:lock (auto-expire TTL) Dequeue uses SCAN to discover queues, checks lock state from scan results (avoiding extra round trips), acquires lock via SET NX, then RPOP. SCAN's hash-table ordering provides natural randomness for fair distribution across partitions. Empty queues are auto-deleted by Redis. Zero keys left behind after all tasks complete. Benchmarked: 6000 tasks across 500 partitions with 8 workers achieved 98% theoretical throughput with 1.5% worker distribution spread. Other changes: - queue_name renamed to queue_prefix (AGENTEXEC_QUEUE_NAME still works) - Removed queue_name parameter from public API (enqueue, dequeue, Pool) - Lock lifecycle owned by queue backend (release_lock on BaseQueueBackend) - Worker no longer handles locks — pool releases on TaskCompleted/TaskFailed - Failed tasks requeued as high priority to preserve execution order - Added examples/queue-fairness/ benchmark 261 passed, 0 failed Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Remove pubsub, inline dequeue, queue.complete, activity over IPC Complete migration of all worker → pool communication to multiprocessing queue. Redis pubsub is fully removed from the system. - Removed publish/subscribe from BaseStateBackend and Redis implementation - Removed _pubsub from Redis Backend (no more pubsub connections) - Activity producer writes create() to Postgres directly (runs on API/pool) - Activity update/complete/error send ActivityUpdated via mp.Queue - Pool handles ActivityUpdated in _process_worker_events match/case - Deleted activity/consumer.py (replaced by inline pool handler) - queue.complete() replaces release_lock() — abstracts lock lifecycle - Worker._run inlines dequeue (pop + validate) and calls complete in finally - Removed dequeue() from core/queue.py (inlined in worker) - Removed _partition_key_for from pool event handler (worker handles it) - Lock methods removed from BaseStateBackend (owned by queue backend) - backend.client property replaces _get_client() method 255 passed, 0 failed Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Schedule backend, session cleanup, dead code removal, resiliency tests - Schedule backend: composite keys (task:cron:hash), Redis hash + sorted set storage - Session management: remove global session, Pool owns engine via configure_engine/get_session - Activity handler pattern: PostgresHandler/IPCHandler with typed events - Remove dead backend methods: configure, index_add/range/remove, clear, publish/subscribe - Remove Kafka pubsub (publish/subscribe) and sorted set cache - Add partition queue tests: SCAN-based dequeue, lock acquisition, multi-partition fairness - Add worker failure tests: TaskFailed IPC, retry with backoff, max retry give-up - Add execute lifecycle tests: None result, TTL storage, context hydration, bad context Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Kafka state: raise NotImplementedError, drop in-memory KV/counter caches Kafka is not a KV store — the per-process caches gave divergent state across workers. State operations now raise NotImplementedError with a clear message. Queue and schedule backends are unaffected. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Fix KafkaStateBackend instantiation (no longer takes backend arg) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Restore docstrings stripped during refactor Args/Returns/Raises blocks, examples, and explanatory comments that were lost when rewriting modules. No behavior changes. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Update Kafka integration tests and fix queue interface mismatch - Remove tests for deleted APIs: state.clear(), activity backend, publish/subscribe, configure(), index_add/range/remove - Add tests for NotImplementedError on state operations - Fix KafkaQueueBackend.push/pop signatures to match BaseQueueBackend (queue_name was an extra arg, now uses CONF.queue_prefix) - Update client_id test for PID-based IDs Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Fix Kafka CI: add OFFSETS_TOPIC_REPLICATION_FACTOR, test timeout Single-node Kafka needs OFFSETS_TOPIC_REPLICATION_FACTOR=1 or consumer groups hang waiting for __consumer_offsets replicas. Also add a 2-minute job timeout and per-test 30s timeout to fail fast instead of hanging. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Remove --timeout flag (pytest-timeout not installed) The job-level timeout-minutes: 2 is sufficient. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Queue fairness benchmark: partition-level metrics, fix stale APIs - Add partition fairness analysis: first-task pickup time, per-partition average wait, starvation detection - Fix stale API calls (push/pop no longer take queue_name, complete replaces release_lock) - Add README documenting benchmark results at 1000 partitions Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Update README for 0.2 API changes and Kafka backend - Document partitioned queue design with SCAN-based fair dequeue - Add Kafka experimental backend section with considerations, limitations, and configuration reference - Update activity API examples to async - Update lock_key docs to describe partition queue routing - Add new config vars to reference (retries, scheduler, Kafka) - Fix stale references throughout Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent 8c4b019 commit db4ec50

55 files changed

Lines changed: 3454 additions & 3607 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

.github/workflows/ci.yml

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
name: CI
2+
3+
on:
4+
push:
5+
branches: [main]
6+
pull_request:
7+
branches: [main]
8+
9+
jobs:
10+
# -----------------------------------------------------------------------
11+
# Unit tests — no external services (fakeredis + SQLite)
12+
# -----------------------------------------------------------------------
13+
14+
test:
15+
runs-on: ubuntu-latest
16+
strategy:
17+
fail-fast: false
18+
matrix:
19+
python-version: ["3.12", "3.13"]
20+
21+
steps:
22+
- uses: actions/checkout@v4
23+
24+
- name: Install uv
25+
uses: astral-sh/setup-uv@v6
26+
with:
27+
enable-cache: true
28+
29+
- name: Set up Python ${{ matrix.python-version }}
30+
run: uv python install ${{ matrix.python-version }}
31+
32+
- name: Install dependencies
33+
run: uv sync --dev
34+
35+
- name: Run unit tests
36+
run: |
37+
uv run pytest tests/ \
38+
--ignore=tests/test_kafka_integration.py \
39+
-o "addopts=" \
40+
-v --tb=long
41+
42+
# -----------------------------------------------------------------------
43+
# Kafka integration tests — real broker via docker run
44+
# -----------------------------------------------------------------------
45+
test-kafka:
46+
runs-on: ubuntu-latest
47+
48+
steps:
49+
- uses: actions/checkout@v4
50+
51+
- name: Start Kafka broker
52+
run: |
53+
docker run -d --name kafka \
54+
-p 9092:9092 \
55+
-e KAFKA_NODE_ID=1 \
56+
-e KAFKA_PROCESS_ROLES=broker,controller \
57+
-e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
58+
-e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
59+
-e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
60+
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
61+
-e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT \
62+
-e KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT \
63+
-e KAFKA_LOG_CLEANER_MIN_COMPACTION_LAG_MS=0 \
64+
-e KAFKA_LOG_CLEANER_MIN_CLEANABLE_RATIO=0.01 \
65+
-e KAFKA_LOG_RETENTION_MS=60000 \
66+
-e KAFKA_NUM_PARTITIONS=1 \
67+
-e KAFKA_AUTO_CREATE_TOPICS_ENABLE=true \
68+
-e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
69+
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
70+
-e CLUSTER_ID=ciTestCluster0001 \
71+
apache/kafka:3.9.0
72+
73+
- name: Install uv
74+
uses: astral-sh/setup-uv@v6
75+
with:
76+
enable-cache: true
77+
78+
- name: Set up Python
79+
run: uv python install 3.12
80+
81+
- name: Install dependencies
82+
run: uv sync --dev --extra kafka
83+
84+
- name: Wait for Kafka to be ready
85+
run: |
86+
echo "Waiting for Kafka..."
87+
for i in $(seq 1 30); do
88+
if nc -z localhost 9092 2>/dev/null; then
89+
echo "Kafka port is open"
90+
sleep 5
91+
echo "Kafka is ready"
92+
exit 0
93+
fi
94+
echo " attempt $i/30..."
95+
sleep 2
96+
done
97+
echo "Kafka failed to start"
98+
docker logs kafka
99+
exit 1
100+
101+
- name: Run Kafka integration tests
102+
timeout-minutes: 2
103+
run: |
104+
uv run pytest tests/test_kafka_integration.py \
105+
-o "addopts=" \
106+
-v --tb=long 2>&1 | tee /tmp/kafka_test_output.txt
107+
exit ${PIPESTATUS[0]}
108+
env:
109+
AGENTEXEC_STATE_BACKEND: agentexec.state.kafka
110+
KAFKA_BOOTSTRAP_SERVERS: localhost:9092
111+
AGENTEXEC_KAFKA_DEFAULT_PARTITIONS: "2"
112+
AGENTEXEC_KAFKA_REPLICATION_FACTOR: "1"
113+
114+
- name: Show Kafka logs on failure
115+
if: failure()
116+
run: docker logs kafka 2>&1 | tail -50
117+
118+
- name: Create failure check annotation with output
119+
if: failure()
120+
run: |
121+
if [ -f /tmp/kafka_test_output.txt ]; then
122+
grep -E '\[queue_|FAILED|ERROR|AssertionError|TIMEOUT|short test summary' /tmp/kafka_test_output.txt | tail -9 | while IFS= read -r line; do
123+
echo "::warning::$line"
124+
done
125+
fi

0 commit comments

Comments
 (0)