Concepts

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:

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

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.

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

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.

@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)

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.

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)

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:

# 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.

Task Statuses

A task's result record carries one of five statuses, surfaced to Ray Serve as a TaskResult(id, status, 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

In v1.0.0 serialization is JSON-only. The task payload sent to the queue is a JSON object:

{
  "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:

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

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:

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

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.

if adapter.health_check():
    print("Broker reachable")

detail = adapter.health_check_sync()
# [{"healthy": True}]

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.

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/ directory for direct QueuesClient, CQClient, and PubSubClient usage, including mixing the adapter with raw SDK calls.

Related Topics