Scenario

An ML platform team runs several models in production at once: a ResNet image classifier, a BERT text embedder, and an LLM completion service. Each model is GPU-bound, has its own throughput profile, and must fail independently — a stuck LLM job should never block image classification.

This scenario wires each model to its own KubeMQ task queue behind a Ray Serve GPU deployment. A front-end HTTP service enqueues work onto the right queue with enqueue_task_sync, GPU consumers process tasks in parallel, long-running jobs stream report_progress events to subscribers, and any task that exhausts its retries fires an on_dlq callback that pages an operator.

Like the multi-language event pipeline, the value here is isolation through queues: each model is an independent consumer of a dedicated channel, so you can scale, version, and A/B-test models without coupling them to each other.

Architecture

A producer enqueues tasks onto model-specific queues. Each model is a @serve.deployment(ray_actor_options={"num_gpus": 1}) consumer that pulls from its queue, writes results to the queue-peek backend, and publishes progress on {queue}.progress. The producer polls get_task_status_sync for the outcome and can subscribe to the progress channel for live updates.

GPU Deployment Binding

In production you do not call initialize() yourself. You decorate a Ray Serve deployment with @serve.deployment to reserve a GPU, then @task_consumer to bind it to a KubeMQ queue through a TaskProcessorConfig. Ray Serve calls initialize() and start_consumer() for you; each method invocation runs as a task pulled off the queue.

image_deployment.py

from ray import serve
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

@serve.deployment(ray_actor_options={"num_gpus": 1})
@task_consumer(
    task_processor_config=TaskProcessorConfig(
        adapter_class=KubeMQTaskProcessorAdapter,
        adapter_config=KubeMQAdapterConfig(
            address="kubemq:50000",
            auth_token="your-jwt-token",
        ),
        queue_name="inference-tasks",
        max_retries=3,
        failed_task_queue_name="inference-tasks-dlq",
    )
)
class ModelDeployment:
    def __init__(self):
        self.model = load_model()

    def __call__(self, data: str) -> dict:
        return self.model.predict(data)

ray_actor_options={"num_gpus": 1} pins one GPU per replica. queue_name is the channel the consumer polls, max_retries caps redelivery attempts before a task is dead-lettered, and failed_task_queue_name is where exhausted tasks land. The auth_token carries a JWT to the in-cluster broker at kubemq:50000.

Each model deployment uses a distinct queue_name. That single decision gives you per-model isolation, independent scaling, and independent failure handling for free — the rest of this page builds on it.

Multiple GPU Models on Separate Queues

Each model gets its own adapter, its own queue, and its own consumer. The producer routes a request to the correct adapter and later collects every result by polling. The snippet below mirrors the packaged gpu_multi_model.py example, which simulates three GPUs (ResNet, BERT, Whisper) on three channels.

gpu_multi_model.py

import os
import time
import uuid

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

BROKER = os.environ.get("KUBEMQ_ADDRESS", "localhost:50000")


def resnet_predict(image_data: str) -> dict:
    """Simulated ResNet image classification (GPU 0)."""
    time.sleep(0.1)  # Simulate 100ms GPU inference
    categories = ["cat", "dog", "bird", "car", "plane"]
    idx = hash(image_data) % len(categories)
    return {"model": "resnet50", "gpu": 0, "prediction": categories[idx]}


def bert_predict(text: str) -> dict:
    """Simulated BERT text classification (GPU 1)."""
    time.sleep(0.12)  # Simulate 120ms GPU inference
    labels = ["positive", "negative", "neutral"]
    return {"model": "bert-base", "gpu": 1, "sentiment": labels[hash(text) % 3]}


def main():
    uid = uuid.uuid4().hex[:8]
    channel_resnet = f"example-ml-gpu-resnet-{uid}"
    channel_bert = f"example-ml-gpu-bert-{uid}"

    # --- Adapter for ResNet (GPU 0) ---
    adapter_resnet = KubeMQTaskProcessorAdapter(KubeMQAdapterConfig(address=BROKER))

    class _CfgResnet:
        queue_name = channel_resnet
        max_retries = 1
        failed_task_queue_name = f"{channel_resnet}.dlq"
        unprocessable_task_queue_name = ""

    adapter_resnet.initialize(consumer_concurrency=2, task_processor_config=_CfgResnet())
    adapter_resnet.register_task_handler(resnet_predict, name="predict")
    adapter_resnet.start_consumer()

    # --- Adapter for BERT (GPU 1) ---
    adapter_bert = KubeMQTaskProcessorAdapter(KubeMQAdapterConfig(address=BROKER))

    class _CfgBert:
        queue_name = channel_bert
        max_retries = 1
        failed_task_queue_name = f"{channel_bert}.dlq"
        unprocessable_task_queue_name = ""

    adapter_bert.initialize(consumer_concurrency=2, task_processor_config=_CfgBert())
    adapter_bert.register_task_handler(bert_predict, name="predict")
    adapter_bert.start_consumer()

    try:
        # Dispatch requests to the appropriate model queue
        img = adapter_resnet.enqueue_task_sync("predict", args=["image_0.jpg"])
        txt = adapter_bert.enqueue_task_sync("predict", args=["Great product!"])

        for adapter, tid in [(adapter_resnet, img.id), (adapter_bert, txt.id)]:
            while True:
                status = adapter.get_task_status_sync(tid)
                if status.status in ("SUCCESS", "FAILURE"):
                    print(status.status, status.result)
                    break
                time.sleep(0.5)
    finally:
        adapter_resnet.stop_consumer()
        adapter_bert.stop_consumer()


if __name__ == "__main__":
    main()

Each handler is registered under the name predict on its own adapter, so the same task name routes to a different model depending on which queue it lands on.

Chained Inference Pipeline

When a single request needs several GPU stages — preprocess, classify, postprocess — give each stage its own queue and chain the calls, passing the result of one stage as the argument to the next. The packaged image_classification_pipeline.py example does exactly this across three adapters.

image_classification_pipeline.py

def poll_result(adapter, task_id, timeout=15):
    start = time.perf_counter()
    while (time.perf_counter() - start) < timeout:
        status = adapter.get_task_status_sync(task_id)
        if status.status in ("SUCCESS", "FAILURE"):
            return {"status": status.status, "result": status.result}
        time.sleep(0.5)
    return None


# Stage 1: Preprocess (resize, normalize) on the preprocess queue
pre_task = adapter_pre.enqueue_task_sync(
    "preprocess", args=["cat_photo.jpg"], kwargs={"target_size": 224}
)
pre = poll_result(adapter_pre, pre_task.id)

# Stage 2: Classify — pass the preprocessed tensor metadata forward
cls_task = adapter_cls.enqueue_task_sync("classify", args=[pre["result"]])
cls = poll_result(adapter_cls, cls_task.id)

# Stage 3: Postprocess — format the final label
post_task = adapter_post.enqueue_task_sync("postprocess", args=[cls["result"]])
final = poll_result(adapter_post, post_task.id)
print(final["result"]["final_label"], final["result"]["confidence_level"])

Splitting stages across queues lets you scale the bottleneck stage independently — for example, run more classifier replicas than preprocessor replicas.

Batch Inference with Parallel Consumers

For high-throughput batch jobs, raise consumer_concurrency so a single deployment processes many tasks in parallel. The producer enqueues the whole batch up front, then drains results as they complete. The packaged batch_inference.py example runs five concurrent consumers.

batch_inference.py

adapter.initialize(consumer_concurrency=5, task_processor_config=_Cfg())
adapter.register_task_handler(classify_item, name="classify_item")
adapter.start_consumer()

# Enqueue the full batch up front
task_ids = []
start = time.perf_counter()
for i in range(BATCH_SIZE):
    result = adapter.enqueue_task_sync("classify_item", args=[i, f"sample text {i}"])
    task_ids.append(result.id)

# Drain results as the parallel consumers complete them
completed = {}
poll_start = time.perf_counter()
while len(completed) < BATCH_SIZE and (time.perf_counter() - poll_start) < 60:
    for tid in task_ids:
        if tid not in completed:
            status = adapter.get_task_status_sync(tid)
            if status.status in ("SUCCESS", "FAILURE"):
                completed[tid] = status
    time.sleep(0.5)

total_time = time.perf_counter() - start
print(f"Throughput: {BATCH_SIZE / total_time:.1f} tasks/sec")

consumer_concurrency controls how many tasks one deployment processes at once. Combine it with the queue-depth autoscaling policy to add replicas as the backlog grows — see the autoscaling guide.

Progress for Long-Running Jobs

LLM completion and text-to-speech jobs run for seconds, not milliseconds. Inside the handler, call report_progress(task_id, pct, detail) to publish progress events on the {queue_name}.progress channel. Subscribers — like the SSE endpoint shown later — receive each update in real time.

The handler needs the task id, so the producer passes it as an explicit task_id kwarg. This pattern comes straight from the packaged llm_inference.py example.

llm_inference.py

adapter: KubeMQTaskProcessorAdapter | None = None  # module-level for handler closure


def generate_completion(prompt: str, max_tokens: int = 50, task_id: str = "") -> dict:
    """Simulated LLM generation with per-step progress reporting."""
    assert adapter is not None

    if task_id:
        adapter.report_progress(task_id, 0, "Tokenizing prompt")

    generated_tokens = []
    for i in range(max_tokens):
        time.sleep(0.02)  # Simulate 20ms per token
        generated_tokens.append(next_token(prompt, i))
        pct = int((i + 1) / max_tokens * 100)
        if (i + 1) % 10 == 0 and task_id:
            adapter.report_progress(task_id, pct, f"Generated {i + 1}/{max_tokens} tokens")

    if task_id:
        adapter.report_progress(task_id, 100, "Generation complete")

    return {
        "prompt": prompt,
        "completion": " ".join(generated_tokens),
        "tokens_generated": max_tokens,
        "finish_reason": "length",
    }


# Producer side: pass an explicit task_id so the handler can report against it
known_task_id = f"llm-task-{uuid.uuid4().hex[:8]}"
result = adapter.enqueue_task_sync(
    "generate",
    kwargs={"prompt": "Explain message queues", "max_tokens": 40, "task_id": known_task_id},
)

Binary Results

Text-to-speech and other models that emit binary output return it base64-encoded so the result stays JSON-safe through the queue-peek backend. The packaged text_to_speech.py example shows the round trip.

text_to_speech.py

import base64


def synthesize_speech(text: str, voice: str = "default", speed: float = 1.0) -> dict:
    audio_bytes = run_tts(text, voice, speed)  # raw WAV bytes
    return {
        "text": text,
        "voice": voice,
        "sample_rate": 22050,
        "audio_format": "wav",
        "audio_size_bytes": len(audio_bytes),
        "audio_base64": base64.b64encode(audio_bytes).decode("ascii"),
    }


# Producer side: decode the base64 payload back into bytes
outcome = poll_result(adapter, result.id)
audio = base64.b64decode(outcome["result"]["audio_base64"])

Operational Concerns

DLQ Alerting

When a task fails and exhausts max_retries, the adapter stores a FAILURE result, nacks the message to the dead letter queue, and fires the on_dlq callback you set on KubeMQAdapterConfig. Use that callback to page an operator the moment a model starts producing permanent failures. The callback receives the task_id and the formatted error string, exactly as the adapter invokes it on retry exhaustion.

dlq_monitoring_callback.py

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter


def on_dlq_callback(task_id: str, error: str) -> None:
    """Called when a task exhausts retries and moves to the DLQ."""
    print(f"[DLQ ALERT] task_id={task_id} error={error}")
    # In production: page on-call, push to PagerDuty / Slack, increment an alert metric


config = KubeMQAdapterConfig(
    address="localhost:50000",
    on_dlq=on_dlq_callback,
)
adapter = KubeMQTaskProcessorAdapter(config)


class _Cfg:
    queue_name = "inference-tasks"
    max_retries = 2
    failed_task_queue_name = "inference-tasks.dlq"
    unprocessable_task_queue_name = ""


adapter.initialize(consumer_concurrency=1, task_processor_config=_Cfg())

The on_dlq callback runs on the consumer thread. Keep it fast and non-blocking — enqueue the alert and return. The adapter wraps the call in a try/except and logs any exception it raises, so a failing alert never crashes the consumer, but it can still slow down processing if it blocks.

Front-End API

Expose enqueue, status, cancel, progress (SSE), and metrics over HTTP so clients never talk to KubeMQ directly. The packaged fastapi_app.py example manages the adapter lifecycle in a FastAPI lifespan and exposes the full surface.

fastapi_app.py

import asyncio
import json
import uuid
from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from kubemq import (
    CancellationToken,
    ClientConfig,
    EventReceived,
    EventsSubscription,
    PubSubClient,
)
from pydantic import BaseModel, Field

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

BROKER = "localhost:50000"
CHANNEL = f"example-fastapi-{uuid.uuid4().hex[:8]}"

config = KubeMQAdapterConfig(address=BROKER)
adapter = KubeMQTaskProcessorAdapter(config)


def echo_handler(text: str = "") -> dict:
    return {"echo": text, "source": "fastapi-app"}


class EnqueueRequest(BaseModel):
    task_name: str
    args: list | None = None
    kwargs: dict | None = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    class _Cfg:
        queue_name = CHANNEL
        max_retries = 3
        failed_task_queue_name = f"{CHANNEL}.dlq"
        unprocessable_task_queue_name = f"{CHANNEL}.unprocessable"

    adapter.initialize(consumer_concurrency=2, task_processor_config=_Cfg())
    adapter.register_task_handler(echo_handler, name="echo")
    adapter.start_consumer()
    yield
    adapter.stop_consumer()


app = FastAPI(title="KubeMQ Ray Serve Example", lifespan=lifespan)


@app.get("/health")
def health():
    return {"healthy": adapter.health_check()}


@app.post("/tasks")
def enqueue_task(req: EnqueueRequest):
    result = adapter.enqueue_task_sync(
        task_name=req.task_name, args=req.args, kwargs=req.kwargs
    )
    return {"task_id": result.id, "status": result.status}


@app.get("/tasks/{task_id}")
def get_task_status(task_id: str):
    status = adapter.get_task_status_sync(task_id)
    return {"task_id": task_id, "status": status.status, "result": status.result}


@app.post("/tasks/{task_id}/cancel")
def cancel_task(task_id: str):
    return {"task_id": task_id, "cancelled": adapter.cancel_task(task_id)}


@app.get("/tasks/{task_id}/progress")
async def task_progress_sse(task_id: str):
    """SSE stream: subscribe to {CHANNEL}.progress, filter by task_id."""
    progress_queue: asyncio.Queue[dict] = asyncio.Queue()
    cancel_token = CancellationToken()
    loop = asyncio.get_running_loop()

    def on_event(event: EventReceived) -> None:
        try:
            data = json.loads(event.body)
            if data.get("task_id") == task_id:
                loop.call_soon_threadsafe(progress_queue.put_nowait, data)
        except (json.JSONDecodeError, TypeError):
            pass

    pubsub = PubSubClient(
        config=ClientConfig(address=BROKER, client_id="fastapi-progress-subscriber")
    )
    subscription = EventsSubscription(
        channel=f"{CHANNEL}.progress",
        on_receive_event_callback=on_event,
        on_error_callback=lambda err: None,
    )
    pubsub.subscribe_to_events(subscription, cancel_token)

    async def event_stream():
        try:
            timeout = 120
            while timeout > 0:
                try:
                    data = await asyncio.wait_for(progress_queue.get(), timeout=5.0)
                    yield f"data: {json.dumps(data)}\n\n"
                    if data.get("pct", 0) >= 100:
                        break
                except asyncio.TimeoutError:
                    yield f"data: {json.dumps({'heartbeat': True})}\n\n"
                timeout -= 5
        finally:
            cancel_token.cancel()
            pubsub.close()

    return StreamingResponse(event_stream(), media_type="text/event-stream")


@app.get("/metrics")
def get_metrics():
    return adapter.get_metrics()

The SSE endpoint subscribes to the same {CHANNEL}.progress channel the handler publishes to, filters events by task_id, and streams data: frames until the job reports 100. The 8-metric get_metrics() dict (queue depth, in-flight, DLQ depth, counters, durations) feeds dashboards and the autoscaler.

flask_app.py

import atexit
import uuid

from flask import Flask, jsonify, request

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

BROKER = "localhost:50000"
CHANNEL = f"example-flask-{uuid.uuid4().hex[:8]}"

config = KubeMQAdapterConfig(address=BROKER)
adapter = KubeMQTaskProcessorAdapter(config)
_adapter_started = False


def echo_handler(text: str = "") -> dict:
    return {"echo": text, "source": "flask-app"}


def _ensure_adapter_started():
    global _adapter_started
    if _adapter_started:
        return

    class _Cfg:
        queue_name = CHANNEL
        max_retries = 3
        failed_task_queue_name = f"{CHANNEL}.dlq"
        unprocessable_task_queue_name = f"{CHANNEL}.unprocessable"

    adapter.initialize(consumer_concurrency=2, task_processor_config=_Cfg())
    adapter.register_task_handler(echo_handler, name="echo")
    adapter.start_consumer()
    _adapter_started = True


app = Flask(__name__)


@app.before_request
def _startup():
    _ensure_adapter_started()


@app.get("/health")
def health():
    return jsonify(healthy=adapter.health_check())


@app.post("/tasks")
def enqueue_task():
    body = request.get_json(force=True)
    result = adapter.enqueue_task_sync(
        task_name=body["task_name"],
        args=body.get("args"),
        kwargs=body.get("kwargs"),
    )
    return jsonify(task_id=result.id, status=result.status)


@app.get("/tasks/<task_id>")
def get_task_status(task_id):
    status = adapter.get_task_status_sync(task_id)
    return jsonify(task_id=task_id, status=status.status, result=status.result)


@app.post("/tasks/<task_id>/cancel")
def cancel_task(task_id):
    return jsonify(task_id=task_id, cancelled=adapter.cancel_task(task_id))


@app.get("/metrics")
def get_metrics():
    return jsonify(adapter.get_metrics())


atexit.register(lambda: adapter.stop_consumer())

Flask exposes the same enqueue, status, cancel, health, and metrics surface, starting the adapter on the first request and stopping it on process exit. For progress SSE, prefer the FastAPI variant — async streaming maps cleanly onto FastAPI's StreamingResponse.

Running the Service

Start KubeMQ

Run the broker locally with both ports exposed:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Port 50000 is the gRPC endpoint the adapters connect to. Port 9090 exposes the shared HTTP server used by connector endpoints such as KEDA-based scaling.

Install dependencies

terminal

uv pip install kubemq-rayserve fastapi uvicorn

Run the front-end API

terminal

python fastapi_app.py

The app serves on http://0.0.0.0:8000. Override the broker address with KUBEMQ_ADDRESS if it is not on localhost:50000.

Enqueue and poll

terminal

curl -s -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"task_name": "echo", "kwargs": {"text": "hello gpu"}}'
# {"task_id":"...","status":"PENDING"}

curl -s http://localhost:8000/tasks/<task_id>
# {"task_id":"...","status":"SUCCESS","result":{"echo":"hello gpu","source":"fastapi-app"}}

Stream progress for a long job with curl -N http://localhost:8000/tasks/<task_id>/progress.

Key Takeaway

One queue per model is the organizing principle. It buys you independent scaling (raise consumer_concurrency and let kubemq_queue_depth_policy add GPU replicas where the backlog is), independent failure handling (on_dlq alerts fire per model), and independent lifecycle (hot-swap or A/B-test a model without touching its neighbors) — all backed by a single KubeMQ broker, with no separate result store or scaler to operate.

GPU Multi-Model Inference Service

Scenario

Architecture

GPU Deployment Binding

Multiple GPU Models on Separate Queues

Chained Inference Pipeline

Batch Inference with Parallel Consumers

Progress for Long-Running Jobs

Binary Results

Operational Concerns

DLQ Alerting

Front-End API

Running the Service

Start KubeMQ

Install dependencies

Run the front-end API

Enqueue and poll

Key Takeaway

Next Steps

Autoscaling Guide

Kubernetes Production Deployment

Progress Tracking

On this page

GPU Multi-Model Inference Service

A/B model testing on separate queues

Hot-swap model versions without downtime

Autoscaling Guide

Kubernetes Production Deployment

Progress Tracking

On this page