Repository: run-llama/llama_deploy

Issue:
There is a conflict between control planes which are reading from the same topic called "control_plane"
There is a significant issue involving a conflict between control planes that are reading from the same Kafka topic named "control_plane." This problem occurs irrespective of the deployment type, whether "API-Server" or "Manual."

Currently, there is no option available to change the name of the topic in the message queue used by the Control Plane server. This limitation results in an odd and undesirable behavior wherein one control plane may wait for a message that has already been consumed by another control plane. Consequently, this leads to timeout errors as the control plane is unable to provide a timely response.

Steps to Reproduce:

Deploy multiple control planes using either "API-Server" or "Manual" deployment methods.
Ensure all control planes are configured to read from the Kafka topic "control_plane."
Observe the behavior of the control planes and note any timeout errors or unexpected message consumption patterns.
Expected Result:

Each control plane should independently consume messages from its own specifically designated topic to prevent conflicts and eliminate timeout errors.

Actual Result:

All control planes read from the same topic, resulting in message consumption conflicts and timeout errors.

Proposed Solution:

Introduce a configuration option that allows users to customize the topic name used by each control plane server to prevent conflicts and ensure smooth communication.

Discussion:
coveralls: 
[![Coverage Status](https://coveralls.io/builds/70803665/badge)](https://coveralls.io/builds/70803665)

coverage: 72.1% (+0.1%) from 71.969%
when pulling **125cf4773f17b29cbbe43c4b1c3a1d9d7372ec6e on massi/fix-349**
into **5f5e76e2c0ba47e644b11e8e2177ffff606f76af on main**.


Patch:
diff --git a/llama_deploy/control_plane/server.py b/llama_deploy/control_plane/server.py
index ba7cd85..289e337 100644
--- a/llama_deploy/control_plane/server.py
+++ b/llama_deploy/control_plane/server.py
@@ -43,17 +43,18 @@ class ControlPlaneConfig(BaseSettings):
         env_prefix="CONTROL_PLANE_", arbitrary_types_allowed=True
     )
 
-    state_store: Optional[BaseKVStore] = None
+    state_store: BaseKVStore | None = None
     services_store_key: str = "services"
     tasks_store_key: str = "tasks"
     session_store_key: str = "sessions"
     step_interval: float = 0.1
     host: str = "127.0.0.1"
-    port: Optional[int] = 8000
-    internal_host: Optional[str] = None
-    internal_port: Optional[int] = None
+    port: int = 8000
+    internal_host: str | None = None
+    internal_port: int | None = None
     running: bool = True
-    cors_origins: Optional[List[str]] = None
+    cors_origins: List[str] | None = None
+    message_type: str = "control_plane"
 
     @property
     def url(self) -> str:
@@ -105,18 +106,19 @@ class ControlPlaneServer(BaseControlPlane):
         self,
         message_queue: BaseMessageQueue,
         orchestrator: BaseOrchestrator,
-        publish_callback: Optional[PublishCallback] = None,
-        state_store: Optional[BaseKVStore] = None,
+        publish_callback: PublishCallback | None = None,
+        state_store: BaseKVStore | None = None,
         services_store_key: str = "services",
         tasks_store_key: str = "tasks",
         session_store_key: str = "sessions",
         step_interval: float = 0.1,
         host: str = "127.0.0.1",
         port: int = 8000,
-        internal_host: Optional[str] = None,
-        internal_port: Optional[int] = None,
+        internal_host: str | None = None,
+        internal_port: int | None = None,
         running: bool = True,
-        cors_origins: Optional[List[str]] = None,
+        cors_origins: List[str] | None = None,
+        message_type: str = "control_plane",
     ) -> None:
         self.orchestrator = orchestrator
 
@@ -136,6 +138,7 @@ class ControlPlaneServer(BaseControlPlane):
         self._message_queue = message_queue
         self._publisher_id = f"{self.__class__.__qualname__}-{uuid.uuid4()}"
         self._publish_callback = publish_callback
+        self._message_type = message_type
 
         self.app = FastAPI()
         if cors_origins:
@@ -294,12 +297,12 @@ class ControlPlaneServer(BaseControlPlane):
                     if self.port
                     else f"http://{self.host}/process_message"
                 ),
-                message_type="control_plane",
+                message_type=self._message_type,
             )
 
         return CallableMessageConsumer(
             id_=self.publisher_id,
-            message_type="control_plane",
+            message_type=self._message_type,
             handler=self.process_message,
         )
 


Test patch:
diff --git a/e2e_tests/message_queues/message_queue_kafka/test_message_queue.py b/e2e_tests/message_queues/message_queue_kafka/test_message_queue.py
index ff7497b..a025939 100644
--- a/e2e_tests/message_queues/message_queue_kafka/test_message_queue.py
+++ b/e2e_tests/message_queues/message_queue_kafka/test_message_queue.py
@@ -22,7 +22,7 @@ def kafka_service():
 
 @pytest.fixture
 def mq(kafka_service):
-    return KafkaMessageQueue(KafkaMessageQueueConfig(topic_name="test"))
+    return KafkaMessageQueue(KafkaMessageQueueConfig(topic_name="test_message"))
 
 
 @pytest.mark.e2e
@@ -35,12 +35,12 @@ async def test_roundtrip(mq):
         received_messages.append(message)
 
     test_consumer = CallableMessageConsumer(
-        message_type="test", handler=message_handler
+        message_type="test_message", handler=message_handler
     )
     start_consuming_callable = await mq.register_consumer(test_consumer)
 
     # produce a message
-    test_message = QueueMessage(type="test", data={"message": "this is a test"})
+    test_message = QueueMessage(type="test_message", data={"message": "this is a test"})
 
     # await asyncio.gather(start_consuming_callable(), mq.publish(test_message))
     await mq.publish(test_message)
diff --git a/tests/control_plane/__init__.py b/tests/control_plane/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/control_plane/test_control_plane.py b/tests/control_plane/test_control_plane.py
new file mode 100644
index 0000000..28582f5
--- /dev/null
+++ b/tests/control_plane/test_control_plane.py
@@ -0,0 +1,22 @@
+from llama_deploy import (
+    ControlPlaneConfig,
+    ControlPlaneServer,
+    SimpleMessageQueue,
+    SimpleMessageQueueConfig,
+    SimpleOrchestrator,
+    SimpleOrchestratorConfig,
+)
+
+
+def test_message_type() -> None:
+    mq = SimpleMessageQueue(**SimpleMessageQueueConfig().model_dump())
+    cfg = ControlPlaneConfig(message_type="foo")
+    control_plane = ControlPlaneServer(
+        mq.client,
+        SimpleOrchestrator(**SimpleOrchestratorConfig().model_dump()),
+        **cfg.model_dump(),
+    )
+    consumer = control_plane.as_consumer(remote=False)
+    assert consumer.message_type == "foo"
+    consumer = control_plane.as_consumer(remote=True)
+    assert consumer.message_type == "foo"