# Concepts (/integrations/rayserve/concepts)
## The TaskProcessorAdapter Framework [#the-taskprocessoradapter-framework]
Ray Serve defines an abstract base class, `TaskProcessorAdapter`, that decouples task distribution from any specific message broker. Ray Serve owns the lifecycle — instantiating the adapter, calling `initialize`, registering handlers, and driving the consumer — while the adapter owns the broker-specific details of enqueuing, consuming, storing results, and reporting health.
`KubeMQTaskProcessorAdapter` is a concrete implementation of that ABC backed by KubeMQ. It implements all 7 abstract methods and 4 optional methods of the framework, plus two extension methods that go beyond what the base class requires:
| Category | Methods |
| ------------- | ------------------------------------------------------------------------------------------------------------------------------------- |
| Abstract (7) | `initialize`, `register_task_handler`, `enqueue_task_sync`, `get_task_status_sync`, `start_consumer`, `stop_consumer`, `health_check` |
| Optional (4) | `cancel_task_sync`, `get_metrics_sync`, plus async-named aliases that delegate to the `_sync` variants |
| Extension (2) | `query_task_sync` (sync inference), `report_progress` (progress events) |
The extension methods are not part of the `TaskProcessorAdapter` contract — they exist because KubeMQ exposes request-response (Queries) and fire-and-forget (Events) primitives that the framework's queue-only model does not. Ray Serve never calls them directly; your handlers and clients do.
## Adapter Lifecycle [#adapter-lifecycle]
The adapter follows a strict five-phase lifecycle. Construction is side-effect-free: `__init__` stores configuration and never touches the network. Connections are created only in `initialize`.
1. **`__init__(config)`** — stores the `KubeMQAdapterConfig` and initializes empty internal state. No client is created yet.
2. **`initialize(consumer_concurrency, task_processor_config=...)`** — builds a `ClientConfig` and creates a single `QueuesClient` shared by both task distribution and result storage, then constructs the `ResultBackend` and `MetricsCollector`. The `task_processor_config` carries `queue_name`, `max_retries`, `failed_task_queue_name`, and `unprocessable_task_queue_name`.
3. **`register_task_handler(func, name)`** — stores the callable under its name (or `func.__name__` if no name is given) in the `_task_handlers` dict.
4. **`start_consumer()`** — sets `_running = True`, creates a `CancellationToken`, launches the daemon polling thread, and starts the Query subscription for sync inference.
5. **`stop_consumer(timeout=10.0)`** — cancels the token, sets `_running = False`, joins the consumer thread, and closes the `QueuesClient` plus any lazily created clients.
```python title="lifecycle.py"
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter
def classify(text: str) -> dict:
return {"label": "POSITIVE", "score": 0.95}
config = KubeMQAdapterConfig(address="localhost:50000")
adapter = KubeMQTaskProcessorAdapter(config) # __init__: no connection
# In production, Ray Serve calls initialize() with a TaskProcessorConfig.
class TaskConfig:
queue_name = "inference-tasks"
max_retries = 3
failed_task_queue_name = "inference-tasks-dlq"
unprocessable_task_queue_name = ""
adapter.initialize(consumer_concurrency=2, task_processor_config=TaskConfig())
adapter.register_task_handler(classify, name="classify")
adapter.start_consumer()
# ... enqueue and poll ...
adapter.stop_consumer()
```
## Threading Model [#threading-model]
The adapter runs two background workers, both started by `start_consumer`.
* **Consumer thread** — a single daemon thread runs `_consumer_loop`, repeatedly calling `receive_queue_messages` with `max_messages=consumer_concurrency`, `wait_timeout_in_seconds=consumer_poll_timeout_seconds`, and `auto_ack=False`. Because auto-ack is disabled, the adapter explicitly acks on success and nacks on failure or shutdown.
* **Query subscription thread** — a separate `CQClient` subscribes to Queries on the *same channel* as the task queue, serving synchronous inference requests through `_handle_sync_query`.
The `QueuesClient` is created eagerly in `initialize`. The other two clients are lazy `cached_property` instances created on first use: `_pubsub_client` (a `PubSubClient` for progress events) and `_cq_client` (a `CQClient` for sync inference). `stop_consumer` only closes the lazy clients if they were ever accessed.
```python title="adapter.py (excerpt)"
@cached_property
def _pubsub_client(self) -> PubSubClient:
"""Lazy PubSubClient for progress events."""
config = self._build_client_config()
return PubSubClient(config=config)
@cached_property
def _cq_client(self) -> CQClient:
"""Lazy CQClient for sync inference."""
config = self._build_client_config()
return CQClient(config=config)
```
The poll loop catches `KubeMQConnectionError` and `KubeMQConnectionNotReadyError`, logs a warning, and sleeps one second before retrying — so a transient broker outage pauses consumption rather than crashing the thread.
## Three Primitives, Three Jobs [#three-primitives-three-jobs]
The adapter maps Ray Serve's task model onto three distinct KubeMQ messaging primitives, each chosen for the delivery semantics it provides.
| KubeMQ Primitive | Client | Job |
| ---------------- | -------------- | ------------------------------------------------------------- |
| Queues | `QueuesClient` | Carry async tasks **and** store results (queue-peek backend) |
| Queries | `CQClient` | Carry synchronous request-response inference |
| Events | `PubSubClient` | Carry progress updates on the `{queue_name}.progress` channel |
Queues give durable, ack-based point-to-point delivery for fire-and-forget inference. Queries give blocking request-response for low-latency synchronous calls. Events give non-durable fan-out for progress notifications that any external subscriber can watch.
## Result Backend (Queue-Peek) [#result-backend-queue-peek]
Results are not stored in a separate Redis or database — they live in KubeMQ itself. The `ResultBackend` writes each task's result as a Queue message on a per-task channel named `{result_channel_prefix}{task_id}` (the default prefix is `rayserve-result-`).
Because a task moves through several states, the backend uses a **purge-then-write** strategy so each state transition overwrites the previous one rather than appending. `store_result` first calls `ack_all_queue_messages` to drain the channel, then sends the new result message:
```python title="result_backend.py (excerpt)"
# Step 1: Purge previous result (state transitions overwrite)
self._client.ack_all_queue_messages(
channel=channel,
wait_time_seconds=_ACK_ALL_PURGE_WAIT_SECONDS,
)
# Step 2: Send new result with retry
for attempt in range(1, _STORE_MAX_RETRIES + 1):
msg = QueueMessage(
channel=channel,
body=body,
expiration_in_seconds=self._expiry,
)
self._client.send_queue_message(msg)
return
```
Retrieval is **non-destructive**: `get_result` uses `peek_queue_messages` so the result stays in the channel and can be polled repeatedly until its `result_expiry_seconds` TTL (default `3600`) lapses. A peek error or a `NOT_FOUND` channel is treated as `PENDING`.
Writes are resilient. `store_result` retries up to **3 times** with a **1-second backoff**, and on persistent failure it falls back to an **in-memory dict capped at 10,000 entries** (evicting the oldest on overflow). Critically, `store_result` **never re-raises** — ML inference is expensive, so losing a result record is preferable to reprocessing a completed task.
The in-memory fallback is per-process. If the adapter process restarts after a fallback write, those results are lost. The fallback is a degraded-mode safety net, not durable storage.
## Task Statuses [#task-statuses]
A task's result record carries one of five statuses, surfaced to Ray Serve as a `TaskResult(id, status, result)`:
| Status | Meaning |
| ----------- | --------------------------------------------------------------------------------------------------------------- |
| `PENDING` | Stored *before* the task is enqueued, so a fast consumer cannot overwrite `SUCCESS` with a late `PENDING` write |
| `STARTED` | Task picked up and processing began |
| `SUCCESS` | Handler returned; `result` holds the return value |
| `FAILURE` | Handler raised after retries were exhausted; `error` holds the description |
| `CANCELLED` | `cancel_task_sync` soft-cancelled the task by overwriting the result |
`PENDING` is deliberately written before the `send_queue_message` call in `enqueue_task_sync` to close a race window — by the time the message reaches a consumer, the `PENDING` record already exists, so a `SUCCESS` write cannot be clobbered by a late `PENDING`.
## Serialization [#serialization]
In v1.0.0 serialization is **JSON-only**. The task payload sent to the queue is a JSON object:
```json title="task payload"
{
"task_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
"task_name": "classify",
"args": ["great product"],
"kwargs": {},
"created_at": "2026-06-01T12:00:00.000000+00:00"
}
```
Arguments that are not JSON-serializable raise a `TypeError` with guidance to convert them first:
```python title="serialization.py (excerpt)"
raise TypeError(
f"Task argument serialization failed: {exc}. "
"Convert complex objects before enqueueing (e.g., ndarray.tolist()). "
"Pluggable serializers planned for v1.1."
) from exc
```
Convert NumPy arrays, tensors, and other non-JSON types before enqueueing — for example, `ndarray.tolist()`. Pluggable serializers are planned for v1.1.
## Retry and DLQ Semantics [#retry-and-dlq-semantics]
When both `max_retries > 0` and `failed_task_queue_name` are configured, `enqueue_task_sync` attaches KubeMQ's native redelivery and dead-letter routing to the queue message:
```python title="adapter.py (excerpt)"
msg_kwargs = {"channel": self._queue_name, "body": body}
if self._max_retries > 0 and self._failed_task_queue_name:
msg_kwargs["max_receive_count"] = self._max_retries
msg_kwargs["max_receive_queue"] = self._failed_task_queue_name
```
On a handler exception, the consumer inspects the message's `receive_count`:
* **Retries remaining** (`receive_count < max_retries`) — the message is **nacked** for redelivery, and the same task is delivered again.
* **Retries exhausted** — the backend stores `FAILURE`, the `on_dlq(task_id, error)` callback fires (if configured), and the nacked message routes to the DLQ via `max_receive_queue`.
A **malformed message** — one whose body is not valid JSON — is handled separately: it is **acked** (so it leaves the queue) and re-sent to `unprocessable_task_queue_name` if that channel is configured. This keeps poison messages from blocking the consumer.
## Health Checks [#health-checks]
Two health methods wrap a single broker `ping()` on the `QueuesClient`:
* **`health_check_sync()`** returns a list with one status dict — `[{"healthy": True}]` on success, or `[{"healthy": False, "error": "..."}]` on failure or before `initialize`.
* **`health_check()`** is a convenience wrapper that reduces that list to a single `bool`.
```python title="health.py"
if adapter.health_check():
print("Broker reachable")
detail = adapter.health_check_sync()
# [{"healthy": True}]
```
## Raw SDK Escape Hatch [#raw-sdk-escape-hatch]
The adapter is built on the standard KubeMQ Python SDK clients (`QueuesClient`, `CQClient`, `PubSubClient`), so power users can mix adapter calls with direct SDK access against the same broker — for example, using the adapter for business logic while inspecting queue depth or peeking result channels directly for operations.
```python title="adapter_plus_raw_sdk.py (excerpt)"
from kubemq import ClientConfig, QueuesClient
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter
# Adapter for task processing
adapter = KubeMQTaskProcessorAdapter(KubeMQAdapterConfig(address=BROKER))
adapter.initialize(consumer_concurrency=1, task_processor_config=cfg)
adapter.register_task_handler(classify, name="classify")
adapter.start_consumer()
# Raw SDK client for monitoring, side by side
sdk_client = QueuesClient(config=ClientConfig(address=BROKER, client_id="raw-monitor"))
channels = sdk_client.list_queues_channels(channel_search="inference-tasks*")
for ch in channels:
print(f"{ch.name}: waiting={ch.incoming.waiting}")
```
See the [`examples/raw_sdk/`](https://github.com/kubemq-io/kubemq-rayserve/tree/main/examples/raw_sdk) directory for direct `QueuesClient`, `CQClient`, and `PubSubClient` usage, including mixing the adapter with raw SDK calls.
## Related Topics [#related-topics]