KubeMQAdapterConfig is a Pydantic BaseModel that carries every connection, result-backend, and consumer-tuning setting the Ray Serve adapter needs. You pass an instance through TaskProcessorConfig.adapter_config when you deploy a @task_consumer, and Ray Serve hands it to KubeMQTaskProcessorAdapter at initialization. Because it is a plain Pydantic model, you build it once, validate it for free, and serialize or diff it like any other config object.

Prerequisites

kubemq-rayserve installed (see Getting Started with Ray Serve)
A running KubeMQ broker reachable at the address you configure below

adapter_config.py

from ray.serve.task_consumer import TaskProcessorConfig
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

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",
)

Field Reference

Every field has a default, so the empty constructor KubeMQAdapterConfig() produces a working configuration against a local broker. Override only what your deployment needs.

Field	Type	Default	Description
`address`	`str`	`"localhost:50000"`	KubeMQ broker address (`host:port`)
`client_id`	`str`	`""`	Client identifier; an auto UUID suffix `rayserve-{8hex}` is generated when empty
`auth_token`	`str`	`""`	JWT authentication token
`tls`	`bool`	`False`	Enable a TLS connection
`tls_cert_file`	`str`	`""`	mTLS client certificate path
`tls_key_file`	`str`	`""`	mTLS client key path
`tls_ca_file`	`str`	`""`	CA certificate path
`result_channel_prefix`	`str`	`"rayserve-result-"`	Prefix for result storage queue channels
`result_expiry_seconds`	`int`	`3600`	Result TTL in seconds (validated `0`–`86400`)
`visibility_timeout`	`int`	`120`	Reserved for future SDK support — not currently wired
`consumer_poll_timeout_seconds`	`int`	`1`	Queue poll wait timeout in seconds
`max_send_size`	`int`	`4194304`	Max message send size in bytes (4 MB)
`max_receive_size`	`int`	`4194304`	Max message receive size in bytes (4 MB)
`sync_inference_timeout`	`int`	`30`	Default timeout for sync Query inference in seconds
`on_dlq`	`Callable`	`None`	Callback `(task_id: str, error: str) -> None` invoked when a task moves to the DLQ

client_id defaults to an empty string. When left empty, the adapter generates rayserve-{8hex} (for example rayserve-3f9 a1c2d) so each replica connects under a distinct identity. Set an explicit client_id only when you need stable, human-readable broker connections.

result_channel_prefix

The queue-peek result backend stores each task's outcome on its own channel named {prefix}{task_id}. Changing the prefix namespaces results so multiple apps, environments, or services can share a broker without colliding on result channels.

result_channel_prefix.py

import os
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

BROKER = os.environ.get("KUBEMQ_ADDRESS", "localhost:50000")
custom_prefix = "my-custom-results-"

config = KubeMQAdapterConfig(
    address=BROKER,
    result_channel_prefix=custom_prefix,
)
adapter = KubeMQTaskProcessorAdapter(config)
# ... initialize, register handlers, start_consumer() ...

result = adapter.enqueue_task_sync("echo", args=["hello"])
print(f"Expected result channel: {custom_prefix}{result.id}")
# Default prefix would produce: rayserve-result-<task_id>

The backend builds the channel name directly from the prefix:

result_backend.py

def _result_channel(self, task_id: str) -> str:
    """Generate result queue channel name for a task."""
    return f"{self._prefix}{task_id}"

consumer_poll_timeout_seconds

This value is the wait_timeout_in_seconds the consumer thread passes to receive_queue_messages. It controls how long each poll blocks waiting for new work before looping again, trading responsiveness against broker call volume.

adapter.py

response = self._client.receive_queue_messages(
    channel=self._queue_name,
    max_messages=self._consumer_concurrency,
    wait_timeout_in_seconds=self._config.consumer_poll_timeout_seconds,
    auto_ack=False,
)

Tune it to the latency profile of your workload:

consumer_poll_timeout.py

from kubemq_rayserve import KubeMQAdapterConfig

# Real-time inference — responsive, slightly higher broker load (default)
realtime = KubeMQAdapterConfig(consumer_poll_timeout_seconds=1)

# Batch / background processing — fewer broker calls, higher latency
batch = KubeMQAdapterConfig(consumer_poll_timeout_seconds=5)

The default of 1 second suits most real-time inference workloads. Raise it to 5–10 seconds for batch or background processing to reduce the number of poll calls against the broker.

max_send_size and max_receive_size

Both default to 4194304 bytes (4 MB) and are applied to the underlying SDK ClientConfig. Raise them when your tasks carry large tensors, images, audio, or documents that exceed the default frame size.

adapter.py

return ClientConfig(
    address=self._config.address,
    client_id=client_id,
    auth_token=self._config.auth_token or None,
    tls=tls_config,
    max_send_size=self._config.max_send_size,
    max_receive_size=self._config.max_receive_size,
)

max_message_size.py

from kubemq_rayserve import KubeMQAdapterConfig

# Default: 4_194_304 (4 MB)
# Custom: 8_388_608 (8 MB) for larger payloads
config = KubeMQAdapterConfig(
    address="localhost:50000",
    max_send_size=8_388_608,    # 8 MB
    max_receive_size=8_388_608,  # 8 MB
)

Keep max_send_size and max_receive_size aligned — a payload that sends successfully must also be receivable on the consumer side. The broker may impose its own maximum message size, so raise the broker limit to match when you increase these values.

sync_inference_timeout

This is the default timeout (in seconds) applied to query_task_sync when a call does not pass its own timeout. The adapter builds the QueryMessage with timeout or self._config.sync_inference_timeout, so a per-call value always wins over the config default.

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)

sync_inference_timeout.py

config = KubeMQAdapterConfig(
    address="localhost:50000",
    sync_inference_timeout=10,  # default is 30
)
adapter = KubeMQTaskProcessorAdapter(config)
# ... initialize, register handlers, start_consumer() ...

# Uses the config default of 10s
result = adapter.query_task_sync("fast_predict", args=[5])

# Per-call timeout=1s overrides the config default
result = adapter.query_task_sync("slow_predict", args=[5], timeout=1)

result_expiry_seconds

The result TTL controls how long a stored result remains readable by get_task_status_sync before the broker expires it. The field is validated to the range 0–86400 seconds (up to 24 hours), defaulting to 3600 (1 hour). Shorten it to reduce broker storage; lengthen it when consumers may poll for results long after completion.

config.py

result_expiry_seconds: int = Field(default=3600, ge=0, le=86400)

result_expiry.py

config = KubeMQAdapterConfig(
    address="localhost:50000",
    result_expiry_seconds=5,  # short-lived results
)
adapter = KubeMQTaskProcessorAdapter(config)
# ... initialize, register handler, start_consumer() ...

result = adapter.enqueue_task_sync("quick_task", args=["test-data"])
status = adapter.get_task_status_sync(result.id)  # SUCCESS, result available

time.sleep(6)  # wait past the 5s TTL
status = adapter.get_task_status_sync(result.id)  # result expired / gone

Choosing a TTL balances result availability against broker storage. A value beyond the validated maximum of 86400 raises a Pydantic validation error at construction time, so misconfiguration fails fast.

All Options at Once

The repository ships an example that sets every field explicitly and then inspects the resulting model. It is the canonical place to see all defaults in one block.

all_config_options.py

from kubemq_rayserve import KubeMQAdapterConfig


def on_dlq_handler(task_id: str, error: str) -> None:
    print(f"DLQ: {task_id}: {error}")


config = KubeMQAdapterConfig(
    # Connection
    address="localhost:50000",
    client_id="my-worker-01",
    auth_token="",
    # TLS / mTLS
    tls=False,
    tls_cert_file="",
    tls_key_file="",
    tls_ca_file="",
    # Result backend
    result_channel_prefix="rayserve-result-",
    result_expiry_seconds=3600,  # 1 hour
    # Consumer tuning
    consumer_poll_timeout_seconds=1,
    max_send_size=4_194_304,  # 4 MB
    max_receive_size=4_194_304,  # 4 MB
    # Sync inference
    sync_inference_timeout=30,
    # Callbacks
    on_dlq=on_dlq_handler,
    # Reserved
    visibility_timeout=120,
)

# Inspect every field and its value
for field_name in KubeMQAdapterConfig.model_fields:
    print(f"  {field_name}: {getattr(config, field_name)!r}")

Environment-Variable-Driven Config

For containers and CI, build the config from environment variables so the same image runs against any broker. Every example honors the KUBEMQ_ADDRESS override, defaulting to localhost:50000 when it is unset.

env_var_config.py

import os
from kubemq_rayserve import KubeMQAdapterConfig

config = KubeMQAdapterConfig(
    address=os.environ.get("KUBEMQ_ADDRESS", "localhost:50000"),
    client_id=os.environ.get("KUBEMQ_CLIENT_ID", ""),
    auth_token=os.environ.get("KUBEMQ_AUTH_TOKEN", ""),
    tls=os.environ.get("KUBEMQ_TLS", "false").lower() == "true",
    tls_ca_file=os.environ.get("KUBEMQ_TLS_CA_FILE", ""),
    tls_cert_file=os.environ.get("KUBEMQ_TLS_CERT_FILE", ""),
    tls_key_file=os.environ.get("KUBEMQ_TLS_KEY_FILE", ""),
)

Run any example against a non-local broker by exporting the override:

terminal

# Default (localhost:50000)
python examples/config/all_config_options.py

# Custom broker address
KUBEMQ_ADDRESS=my-broker:50000 python examples/config/all_config_options.py

If you do not yet have a broker, start one locally with Docker. 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.

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

Serialization Notes

The config model is safe to serialize even though it carries a callback — auth_token never appears in repr() or log output, and on_dlq is dropped from model_dump(). See Validation and serialization in the configuration reference for the full field-level explanation.

Configuration Reference for every KubeMQAdapterConfig field, validation rules, and dependencies
Getting Started with Ray Serve for installing the adapter and running a first task
Ray Serve integration overview for architecture and inference models

Configuration

On this page