✨ Kafka server by aleksul · Pull Request #229 · aleksul/repid

aleksul · 2026-04-10T11:08:46Z

Change Summary

Kafka server

Related issue number

Fix #98

Checklist

Unit tests for the changes exist
Tests pass on CI and coverage remains at 100%
Documentation reflects the changes where applicable
My PR is ready to review

github-actions · 2026-04-10T11:11:51Z

Coverage Report

Name	Stmts	Miss	Cover	Missing
TOTAL	7094	0	100%

96 files skipped due to complete coverage.

cloudflare-workers-and-pages · 2026-04-11T00:05:33Z

Deploying repid with Cloudflare Pages

Latest commit:	`aa164a1`
Status:	✅ Deploy successful!
Preview URL:	https://b766a750.repid.pages.dev
Branch Preview URL:	https://kafka.repid.pages.dev

View logs

Copilot

Pull request overview

Adds a Kafka (Redpanda-backed) server implementation to Repid to support publish/subscribe and message acknowledgements via the existing Repid server APIs (Fix #98).

Changes:

Introduces KafkaServer, KafkaSubscriber, and KafkaReceivedMessage implementations using aiokafka.
Adds Redpanda-based integration fixtures and Kafka-specific integration tests covering ack/nack/reject/reply and subscriber lifecycle.
Adds an optional kafka extra (aiokafka) and updates mypy/pre-commit configuration to accommodate the new dependency.

Reviewed changes

Copilot reviewed 9 out of 10 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
`repid/connections/kafka/message_broker.py`	New Kafka server implementation (publish/subscribe, connection lifecycle).
`repid/connections/kafka/subscriber.py`	New consumer/subscriber loop with concurrency limiting and manual commits.
`repid/connections/kafka/message.py`	New received-message wrapper implementing ack/nack/reject/reply semantics.
`repid/connections/kafka/__init__.py`	Exposes `KafkaServer` from the kafka package.
`repid/connections/__init__.py`	Conditionally exports `KafkaServer` when `aiokafka` is installed.
`tests/integration/conftest.py`	Adds Redpanda container + `kafka_connection` fixture; includes Kafka in integration parametrization.
`tests/integration/test_kafka_specific.py`	Adds Kafka-specific integration tests for message actions and subscriber lifecycle.
`pyproject.toml`	Adds `kafka` optional dependency and mypy override for `aiokafka.*`.
`.pre-commit-config.yaml`	Adjusts mypy pre-commit entry invocation.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

repid/connections/kafka/subscriber.py

repid/connections/kafka/message_broker.py

repid/connections/kafka/message.py

Copilot

Pull request overview

Copilot reviewed 10 out of 11 changed files in this pull request and generated 7 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

tests/integration/conftest.py

repid/connections/kafka/subscriber.py

repid/connections/kafka/message.py

Copilot

Pull request overview

Copilot reviewed 10 out of 11 changed files in this pull request and generated 4 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

repid/connections/kafka/subscriber.py

repid/connections/kafka/message_broker.py

tests/integration/conftest.py

Copilot

Pull request overview

Copilot reviewed 10 out of 11 changed files in this pull request and generated 5 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

repid/connections/kafka/message.py

repid/connections/kafka/subscriber.py

tests/integration/conftest.py

Copilot

Pull request overview

Copilot reviewed 10 out of 11 changed files in this pull request and generated 4 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

tests/integration/conftest.py

repid/connections/kafka/message_broker.py

repid/connections/kafka/subscriber.py

Copilot

Pull request overview

Copilot reviewed 10 out of 11 changed files in this pull request and generated 6 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-04-12T01:15:30Z

tests/integration/test_kafka_specific.py

+        await asyncio.wait_for(event.wait(), timeout=10.0)
+        await subscriber.close()
+
+        assert received_msg is not None
+        assert received_msg.payload == b"reject_payload"
+        assert received_msg.content_type == "text/plain"
+
+        await received_msg.reject()


In this test the subscriber is closed before calling received_msg.reject(). KafkaReceivedMessage.reject() ultimately calls the subscriber’s consumer.commit(...) via the mark-complete callback, but KafkaSubscriber.close() stops the consumer, so committing after close() is likely to fail/flap. Keep the subscriber open until after the message has been acted on (reject/nack/reply), or adjust the Kafka implementation so message actions can still commit after subscriber shutdown.

Suggested change

await asyncio.wait_for(event.wait(), timeout=10.0)

await subscriber.close()

assert received_msg is not None

assert received_msg.payload == b"reject_payload"

assert received_msg.content_type == "text/plain"

await received_msg.reject()

try:

await asyncio.wait_for(event.wait(), timeout=10.0)

assert received_msg is not None

assert received_msg.payload == b"reject_payload"

assert received_msg.content_type == "text/plain"

await received_msg.reject()

finally:

await subscriber.close()

Copilot · 2026-04-12T01:15:31Z

tests/integration/test_kafka_specific.py

+        await subscriber.close()
+
+        assert received_msg is not None
+
+        assert received_msg.content_type == "application/json"
+        await received_msg.nack()


Same issue as above: the subscriber is closed before received_msg.nack() is called. Since nack commits the offset via the consumer, calling it after subscriber.close() (which stops the consumer) is likely to fail/flap. Move await subscriber.close() to after the nack (and any follow-up assertions that require the action), or change the Kafka implementation to allow committing after close.

Suggested change

await subscriber.close()

assert received_msg is not None

assert received_msg.content_type == "application/json"

await received_msg.nack()

assert received_msg is not None

assert received_msg.content_type == "application/json"

await received_msg.nack()

await subscriber.close()

Copilot · 2026-04-12T01:15:31Z

tests/integration/test_kafka_specific.py

+        await asyncio.wait_for(event.wait(), timeout=10.0)
+        await subscriber.close()
+
+        assert received_msg is not None
+
+        await received_msg.reply(
+            payload=b"reply_response",
+            headers={"is_reply": "true"},
+            content_type="application/json",
+            channel=reply_channel_name,
+        )
+


Same ordering issue: the subscriber is closed before received_msg.reply(...) is invoked, but reply commits via the consumer. With the consumer stopped in subscriber.close(), the commit can fail. Call reply(...) before closing the subscriber (or make Kafka message actions independent of subscriber shutdown).

Copilot · 2026-04-12T01:15:31Z

tests/integration/conftest.py

+    for _ in range(100):
+        try:
+            kafka_container._container.reload()
+            port = kafka_container.ports["29092/tcp"][0]
+            break
+        except (KeyError, AttributeError):
+            sleep(0.1)


This is an async fixture but it calls time.sleep(...) while waiting for the container port, which blocks the event loop and can slow/flap the entire async test session. Use await asyncio.sleep(...) in this loop (and consider using a monotonic deadline rather than a fixed iteration count).

Suggested change

for _ in range(100):

try:

kafka_container._container.reload()

port = kafka_container.ports["29092/tcp"][0]

break

except (KeyError, AttributeError):

sleep(0.1)

loop = asyncio.get_running_loop()

deadline = loop.time() + 10

while loop.time() < deadline:

try:

kafka_container._container.reload()

port = kafka_container.ports["29092/tcp"][0]

break

except (KeyError, AttributeError):

await asyncio.sleep(0.1)

Copilot · 2026-04-12T01:15:31Z

repid/connections/kafka/message_broker.py

+            "supports_acknowledgments": True,
+            "supports_persistence": True,
+            "supports_reply": True,
+            "supports_lightweight_pause": False,


supports_lightweight_pause is set to False, but KafkaSubscriber.pause()/resume() uses Kafka’s native consumer.pause()/resume() and does not require disconnecting/re-subscribing. Runner uses this capability flag for backpressure (repid/_runner.py:291+), so marking it false will prevent pausing even though it appears supported. Consider setting supports_lightweight_pause to True (and updating the corresponding integration assertion).

Suggested change

"supports_lightweight_pause": False,

"supports_lightweight_pause": True,

Copilot · 2026-04-12T01:15:31Z

repid/connections/kafka/subscriber.py

+                        self._background_tasks.add(task)
+                        task.add_done_callback(self._background_tasks.discard)
+        except asyncio.CancelledError:
+            pass


_consume_loop() only suppresses CancelledError. If consumer.getmany() (or iteration/processing) raises any other exception (e.g. connection issues), the task will exit and the consumer will never be stopped, potentially leaking resources and leaving the subscriber silently inactive. Handle unexpected exceptions (log + retry with backoff, or transition to a closed state and stop the consumer in a finally).

Suggested change

pass

self._closed = True

raise

except Exception as exc:

self._closed = True

logger.exception("subscriber.consume_loop.error", exc_info=exc)

finally:

self._closed = True

with contextlib.suppress(Exception):

await self._consumer.stop()

aleksul force-pushed the kafka branch from 5b293a7 to 76b5d30 Compare April 11, 2026 00:05

aleksul force-pushed the kafka branch from 76b5d30 to e621ea8 Compare April 11, 2026 00:07

aleksul requested a review from Copilot April 11, 2026 12:05

Copilot started reviewing on behalf of aleksul April 11, 2026 12:06 View session

Copilot AI reviewed Apr 11, 2026

View reviewed changes

aleksul force-pushed the kafka branch 2 times, most recently from a768109 to 39402c3 Compare April 11, 2026 18:55

aleksul requested a review from Copilot April 11, 2026 18:55

Copilot started reviewing on behalf of aleksul April 11, 2026 18:56 View session

Copilot AI reviewed Apr 11, 2026

View reviewed changes

aleksul force-pushed the kafka branch from 39402c3 to a26c224 Compare April 11, 2026 23:38

aleksul requested a review from Copilot April 11, 2026 23:38

Copilot started reviewing on behalf of aleksul April 11, 2026 23:39 View session

Copilot AI reviewed Apr 11, 2026

View reviewed changes

repid/connections/kafka/subscriber.py Outdated Show resolved Hide resolved

repid/connections/kafka/message_broker.py Outdated Show resolved Hide resolved

repid/connections/kafka/message_broker.py Show resolved Hide resolved

tests/integration/conftest.py Outdated Show resolved Hide resolved

aleksul force-pushed the kafka branch from a26c224 to cb9a639 Compare April 12, 2026 00:42

aleksul requested a review from Copilot April 12, 2026 00:42

Copilot started reviewing on behalf of aleksul April 12, 2026 00:42 View session

Copilot AI reviewed Apr 12, 2026

View reviewed changes

aleksul force-pushed the kafka branch from cb9a639 to c4e39cb Compare April 12, 2026 00:58

aleksul requested a review from Copilot April 12, 2026 00:59

Copilot started reviewing on behalf of aleksul April 12, 2026 00:59 View session

Copilot AI reviewed Apr 12, 2026

View reviewed changes

tests/integration/conftest.py Show resolved Hide resolved

tests/integration/conftest.py Outdated Show resolved Hide resolved

repid/connections/kafka/message_broker.py Show resolved Hide resolved

repid/connections/kafka/subscriber.py Show resolved Hide resolved

aleksul force-pushed the kafka branch from c4e39cb to a4003e2 Compare April 12, 2026 01:09

aleksul requested a review from Copilot April 12, 2026 01:09

Copilot started reviewing on behalf of aleksul April 12, 2026 01:10 View session

Copilot AI reviewed Apr 12, 2026

View reviewed changes

✨ Kafka server

aa164a1

aleksul force-pushed the kafka branch from a4003e2 to aa164a1 Compare April 12, 2026 01:18

aleksul merged commit 166411d into main Apr 12, 2026
13 checks passed

aleksul deleted the kafka branch April 12, 2026 01:22

	"supports_lightweight_pause": False,
	"supports_lightweight_pause": True,

-            pass
+            self._closed = True
+            raise
+        except Exception as exc:
+            self._closed = True
+            logger.exception("subscriber.consume_loop.error", exc_info=exc)
+        finally:
+            self._closed = True
+            with contextlib.suppress(Exception):
+                await self._consumer.stop()

Uh oh!

Conversation

aleksul commented Apr 10, 2026

Change Summary

Related issue number

Checklist

Uh oh!

github-actions bot commented Apr 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Coverage Report

Uh oh!

cloudflare-workers-and-pages bot commented Apr 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Deploying repid with Cloudflare Pages

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI Apr 12, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 12, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 12, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Apr 12, 2026

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented Apr 10, 2026 •

edited

Loading

cloudflare-workers-and-pages bot commented Apr 11, 2026 •

edited

Loading