feat(BA-903): Implement Kafka Message Queue #3922
Conversation
github-actions
bot
added
size:XL
| raise ValueError(f"Unsupported serialization format: {format}") | ||
|
|
||
| @classmethod | ||
| def deserialize(cls, data: bytes, format: str = "msgpack") -> Self: |
There was a problem hiding this comment.
Consider using typing.Literal here for better type safety and readability. See the official documentation for more details: https://docs.python.org/3/library/typing.html#typing.Literal.
| def deserialize(cls, data: bytes, format: str = "msgpack") -> Self: | |
| def deserialize(cls, data: bytes, format: Literal["msgpack", "json"] = "msgpack") -> Self: |
There was a problem hiding this comment.
ty for the feedback!
src/ai/backend/common/events.py
Outdated
| async with aclosing( | ||
| redis_helper.read_stream_by_group( | ||
| self.redis_client, | ||
| await self.message_queue.receive_group( | ||
| self._stream_key, | ||
| self._consumer_group, | ||
| self._consumer_name, | ||
| ) | ||
| ) as agen: |
There was a problem hiding this comment.
It would be nice if aclosing was also handled inside the message queue.
tests/common/test_events.py
Outdated
| @@ -132,7 +140,7 @@ def scb(context: object, source: AgentId, event: DummyEvent) -> None: | |||
| assert "ZeroDivisionError" in exception_log | |||
| assert "OverflowError" in exception_log | |||
|
|
|||
| await redis_helper.execute(producer.redis_client, lambda r: r.flushdb()) | |||
| await redis_helper.execute(producer.message_queue.connection_info, lambda r: r.flushdb()) | |||
There was a problem hiding this comment.
If possible, avoid logic that passes internal fields to external.
| T = TypeVar("T", bound=BaseConnectionInfo) | ||
|
|
||
| class AbstractMessageQueue(ABC, Generic[T]): | ||
| connection_info: T |
There was a problem hiding this comment.
Why should connection_info be a common field?
There was a problem hiding this comment.
connection_info is accessed as a field, should i remove it from being a common field and have a get function to return it instead?
There was a problem hiding this comment.
My guess is that each sub class have its own connection_info, and it doesn't seem necessary to have it as a common field in ABC.
| task_id: Final[uuid.UUID] | ||
| total_progress: Union[int, float] | ||
| current_progress: Union[int, float] | ||
|
|
||
| def __init__( | ||
| self, | ||
| # redis_config: RedisConfig, |
There was a problem hiding this comment.
redis_connection is not set up in constructor
resolves #3887 (BA-903)
About
Checklist: (if applicable)
ai.backend.testdocsdirectory