# Configuration (/integrations/rayserve/how-to/configuration)



`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 [#prerequisites]

* `kubemq-rayserve` installed (see [Getting Started with Ray Serve](/integrations/rayserve/tutorials/getting-started))
* A running KubeMQ broker reachable at the `address` you configure below

```python title="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 [#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 |

<Callout type="info">
  `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.
</Callout>

## result\_channel\_prefix [#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.

```python title="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:

```python title="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 [#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.

```python title="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:

```python title="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)
```

<Callout type="info">
  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.
</Callout>

## max\_send\_size and max\_receive\_size [#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.

```python title="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,
)
```

```python title="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
)
```

<Callout type="warn">
  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.
</Callout>

## sync\_inference\_timeout [#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.

```python title="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)
```

```python title="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 [#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.

```python title="config.py"
result_expiry_seconds: int = Field(default=3600, ge=0, le=86400)
```

```python title="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
```

<Callout type="info">
  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.
</Callout>

## All Options at Once [#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.

```python title="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 [#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.

```python title="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:

```bash title="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.

<RunKubeMQ ports="[50000, 9090]" />

## Serialization Notes [#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](/integrations/rayserve/reference/configuration#validation-and-serialization)
in the configuration reference for the full field-level explanation.

## Related [#related]

* [Configuration Reference](/integrations/rayserve/reference/configuration) for every `KubeMQAdapterConfig` field, validation rules, and dependencies
* [Getting Started with Ray Serve](/integrations/rayserve/tutorials/getting-started) for installing the adapter and running a first task
* [Ray Serve integration overview](/integrations/rayserve) for architecture and inference models