Overview

Sync inference is for low-latency, request-response calls where the caller blocks until the result is ready, instead of enqueueing a task and polling for it. It is built on KubeMQ Queries through the SDK CQClient — a feature the Celery TaskProcessorAdapter does not offer.

Use sync inference when:

A single request needs an immediate answer (an HTTP handler waiting on a model prediction, an interactive UI, a synchronous RPC).
The expected handler duration is short enough to wait on inline.
You want to avoid the polling-interval overhead that async inference incurs.

Prefer async inference for fire-and-forget submission, batch workloads, or long-running jobs where blocking a caller is wasteful.

Sync inference runs over KubeMQ Queries, a separate channel from the async task queue. The same start_consumer() call serves both paths, so a single deployed adapter handles enqueued tasks and sync queries concurrently.

The following diagram shows the sync path. query_task_sync sends a Query through the CQClient; the query subscription thread on the server dispatches it to the registered handler and returns the response in line — no queue, no result backend, no polling.

How It Works

The `query_task_sync` method

query_task_sync(task_name, args, kwargs, timeout) serializes a sync query payload of the form {task_name, args, kwargs}, sends a QueryMessage with timeout_in_seconds, blocks until the response or timeout, and returns a TaskResult(status, result).

signature

def query_task_sync(
    self,
    task_name: str,
    args: Any = None,
    kwargs: Any = None,
    timeout: int = 30,
    **options: Any,
) -> TaskResult:
    ...

The wire payload contains only task_name, args, and kwargs — no task_id and no created_at, because the response travels back through the Query channel rather than being stored.

sync query payload

{
  "task_name": "greet",
  "args": ["World"],
  "kwargs": {}
}

Server side

start_consumer() starts the queue-polling consumer thread and a Query subscription. The subscription opens a QueriesSubscription on the adapter's queue_name and routes every incoming query through _handle_sync_query, which:

Deserializes the query body into task_name, args, and kwargs.
Looks up the registered handler by task_name.
Calls handler(*args, **kwargs) and serializes the return value as a SUCCESS response.
Sends a QueryResponse back through the CQClient.

If the handler name is unknown, or the handler raises, the response carries status="FAILURE" and an error string instead of crashing the subscription:

server-side dispatch (adapter.py)

handler = self._task_handlers.get(task_name)
if handler is None:
    response_body = serialize_sync_response(
        "FAILURE", error=f"Unknown task handler: {task_name}"
    )
else:
    try:
        result = handler(*args, **kwargs)
        response_body = serialize_sync_response("SUCCESS", result=result)
    except Exception as exc:
        response_body = serialize_sync_response(
            "FAILURE", error=f"{type(exc).__name__}: {exc}"
        )

response = QueryResponse(
    query_received=query,
    body=response_body,
    is_executed=True,
)
self._cq_client.send_response_message(response)

Sync responses are not stored in the result backend. The handler's return value is encoded directly into the QueryResponse body and returned in line through the Query channel — so there is nothing to poll and nothing to expire. This is the key difference from async inference, where results are written to the queue-peek backend and read back with get_task_status_sync.

Prerequisites

Sync inference needs a running KubeMQ broker (default address localhost:50000). Start one locally with Docker:

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 KubeMQ's gRPC endpoint used by the adapter's SDK clients. Port 9090 exposes the shared HTTP server used by connector endpoints such as KEDA scaling. Install the adapter:

terminal

uv pip install kubemq-rayserve

Minimal Example

Register a greet handler, send one sync query with query_task_sync, and print the result. No polling loop is needed — the call blocks and returns the answer directly.

sync_hello_world.py

from __future__ import annotations

import os
import uuid

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

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


def greet(name: str) -> dict:
    return {"greeting": f"Hello, {name}!"}


def main():
    channel = f"example-quickstart-{uuid.uuid4().hex[:8]}"

    config = KubeMQAdapterConfig(address=BROKER, sync_inference_timeout=10)
    adapter = KubeMQTaskProcessorAdapter(config)

    class _Cfg:
        queue_name = channel
        max_retries = 0
        failed_task_queue_name = ""
        unprocessable_task_queue_name = ""

    adapter.initialize(consumer_concurrency=1, task_processor_config=_Cfg())
    adapter.register_task_handler(greet, name="greet")
    adapter.start_consumer()

    try:
        print("Sending sync inference query...")
        result = adapter.query_task_sync("greet", args=["World"], timeout=10)
        print(f"  status={result.status}")
        print(f"  result={result.result}")
    finally:
        adapter.stop_consumer()

    print("Example complete.")


if __name__ == "__main__":
    main()

Expected output:

output

Sending sync inference query...
  status=SUCCESS
  result={'greeting': 'Hello, World!'}
Example complete.

In production you deploy the adapter inside a Ray Serve @task_consumer and call query_task_sync from your request path. The standalone setup above (initialize + register_task_handler + start_consumer) is used here so the example runs as a single script against a local broker.

Timeout Behavior

The timeout argument is the per-call query deadline in seconds. When you pass a value it overrides config.sync_inference_timeout (default 30); when omitted or falsy it falls back to that config value:

adapter.py

query = QueryMessage(
    channel=self._queue_name,
    body=body,
    timeout_in_seconds=timeout or self._config.sync_inference_timeout,
)
response = self._cq_client.send_query(query)

If no response arrives before the deadline, the SDK raises KubeMQTimeoutError. Catch it on the caller side and decide whether to retry, fail the request, or fall back to async submission:

timeout_handling.py

from kubemq.core.exceptions import KubeMQTimeoutError


def slow_handler(data: str) -> dict:
    import time
    time.sleep(5)  # longer than the query timeout below
    return {"data": data, "processed": True}


# slow_handler takes 5s, but the query deadline is only 1s
try:
    result = adapter.query_task_sync("slow_handler", args=["slow-data"], timeout=1)
    print(f"Unexpected success: {result.result}")
except KubeMQTimeoutError as exc:
    print(f"Timed out: {exc}")
    # The handler needs more time than the query allows — retry with a
    # larger timeout, or enqueue it asynchronously instead.

Set the timeout to your expected handler duration plus a margin. As a rule of thumb: fast handlers timeout=5-10, slower handlers timeout=30-60. If a handler routinely needs more than a minute, use async inference instead of blocking the caller.

Sync vs Async Tradeoffs

The same task can run either way. Sync blocks the caller and returns the result inline; async enqueues the task and polls the result backend until it completes.

	Sync (`query_task_sync`)	Async (`enqueue_task_sync` + poll)
Transport	KubeMQ Query (`CQClient`)	KubeMQ Queue (`QueuesClient`)
Caller	Blocks until result or timeout	Returns immediately, polls later
Result delivery	In line through the Query channel	Stored in the queue-peek result backend
Polling overhead	None	Bounded by the poll interval
Best for	Interactive, single, low-latency requests	Fire-and-forget, batch, long-running jobs
Retries / DLQ	None — failure returns `FAILURE`	`max_retries`, `failed_task_queue_name`, `on_dlq`

The comparison example runs both paths against the same handler and prints their latencies side by side:

sync_vs_async_comparison.py

input_text = "The quick brown fox jumps over the lazy dog"

# --- Sync approach: query_task_sync ---
t0 = time.time()
sync_result = adapter.query_task_sync("summarize", args=[input_text], timeout=10)
sync_elapsed = time.time() - t0
print(f"Sync:  status={sync_result.status} elapsed={sync_elapsed:.3f}s")

# --- Async approach: enqueue + poll ---
t0 = time.time()
enqueued = adapter.enqueue_task_sync("summarize", args=[input_text])
for _ in range(30):
    status = adapter.get_task_status_sync(enqueued.id)
    if status.status in ("SUCCESS", "FAILURE"):
        break
    time.sleep(0.5)
async_elapsed = time.time() - t0
print(f"Async: status={status.status} elapsed={async_elapsed:.3f}s")

Sync is typically faster for single requests because it avoids the polling-interval overhead; async is the better fit for fire-and-forget or batch workloads where the caller should not block.

Multiple Handlers Over Sync

A single adapter can register many handlers and dispatch sync queries to each by task_name. Every handler shares the same Query channel; the task_name in the payload selects which one runs.

sync_multiple_handlers.py

def sentiment(text: str) -> dict:
    positive_words = {"good", "great", "excellent", "happy", "love"}
    words = set(text.lower().split())
    score = len(words & positive_words) / max(len(words), 1)
    return {"label": "positive" if score > 0.2 else "neutral", "score": round(score, 2)}


def tokenize(text: str) -> dict:
    tokens = text.split()
    return {"tokens": tokens, "count": len(tokens)}


def language_detect(text: str) -> dict:
    spanish = {"el", "la", "de", "en", "es", "hola"}
    words = set(text.lower().split())
    return {"language": "es" if words & spanish else "en", "confidence": 0.95}


adapter.register_task_handler(sentiment, name="sentiment")
adapter.register_task_handler(tokenize, name="tokenize")
adapter.register_task_handler(language_detect, name="language_detect")
adapter.start_consumer()

text = "This is a great and excellent product"

# Dispatch the same input to three different handlers by name.
for task in ("sentiment", "tokenize", "language_detect"):
    result = adapter.query_task_sync(task, args=[text], timeout=10)
    print(f"{task}: status={result.status} result={result.result}")

A query whose task_name does not match any registered handler returns a TaskResult with status="FAILURE" and an Unknown task handler error — it never blocks indefinitely.

Sync Inference

Overview

How It Works

The `query_task_sync` method

Server side

Prerequisites

Minimal Example

Timeout Behavior

Sync vs Async Tradeoffs

Multiple Handlers Over Sync

Next Steps

Async Inference

API Reference

On this page