[serve] [taskiq - 2/n] Add multi-broker Taskiq adapter config and initialization

[harshit-anyscale]

Summary

• Adds TaskiqAdapterConfig and TaskiqTaskProcessorAdapter (config + initialization only) for async task processing in Ray Serve
• Full adapter implementation (enqueue, consume, shutdown) coming in follow-up PR

Design: Why broker_type + broker_kwargs

We evaluated several approaches for supporting multiple Taskiq brokers. In taskiq, for each broker, there is a separate class that we need to initialize, so somewhere we need to identify the broker type and accordingly call that broker's class in taskiq:

Approaches considered

1. Per-broker config classes with discriminated union — A typed Pydantic config class per broker (e.g., RedisStreamBrokerConfig, RabbitMQBrokerConfig) with create_broker() factory methods, combined via a Pydantic discriminated union.

2. broker_type + broker_url + broker_kwargs — A single flat config with the connection URL as a top-level field and everything else passed through a dict.

3. broker_type + broker_kwargs (chosen) — A single flat config where all broker constructor arguments (including connection URLs) are passed through broker_kwargs.

Why we chose approach 3

Per-broker config classes (approach 1) had a maintenance problem:

• Each broker has a unique constructor signature with different parameter names, types, and defaults
• We'd need to mirror every broker's defaults accurately in their respective adapter config classes
• Every upstream broker update could silently break our config classes
• Taskiq-redis alone has 9 broker classes — maintaining typed configs for all of them is not scalable

Separating broker_url (approach 2) didn't work uniformly:

• Different brokers use different parameter names for the connection: url (Redis, RabbitMQ), servers (NATS), bootstrap_servers (Kafka)
• Redis Sentinel brokers don't take a URL at all — they require sentinels: list[tuple[str, int]] + master_name: str
• This forced us to maintain a URL-to-param mapping dict and still couldn't handle Sentinel

The broker_kwargs pass-through (approach 3) is the simplest:

• Follows the same pattern as CeleryAdapterConfig (which uses broker_transport_options: Dict[str, Any] for pass-through)
• Broker handles its own defaults — no mismatch risk
• Adding a new broker type = one registry entry (import path + package name + queue param)

What we validate

Even with pass-through kwargs, we validate at our layer:

• broker_type must be a known registry key
• Required kwargs per broker are checked before calling the constructor (e.g., redis_stream requires url, redis_stream_sentinel requires sentinels + master_name)
• Lazy imports with helpful error messages if the broker package isn't installed

Config Structure

class TaskiqAdapterConfig(BaseModel):
    broker_type: str                          # required — e.g. "redis_stream", "rabbitmq", "nats"
    broker_kwargs: Optional[Dict[str, Any]]   # pass-through to broker constructor
    result_backend_url: Optional[str]         # Redis URL for result backend

Usage Examples

# Redis Streams
TaskiqAdapterConfig(
    broker_type="redis_stream",
    broker_kwargs={"url": "redis://localhost:6379"},
)

# Redis Sentinel
TaskiqAdapterConfig(
    broker_type="redis_stream_sentinel",
    broker_kwargs={
        "sentinels": [("sentinel1", 26379), ("sentinel2", 26379)],
        "master_name": "mymaster",
    },
)

# RabbitMQ
TaskiqAdapterConfig(
    broker_type="rabbitmq",
    broker_kwargs={"url": "amqp://guest:guest@localhost:5672", "qos": 10},
)

# NATS (multi-server)
TaskiqAdapterConfig(
    broker_type="nats",
    broker_kwargs={"servers": ["nats://host1:4222", "nats://host2:4222"]},
)

# Kafka
TaskiqAdapterConfig(
    broker_type="kafka",
    broker_kwargs={"bootstrap_servers": ["localhost:9092"]},
)

What's in this PR

• TaskiqAdapterConfig — config class with broker_type, broker_kwargs, result_backend_url
• TaskiqTaskProcessorAdapter.__init__() + initialize() — creates broker and optional result backend
• Abstract method stubs for remaining adapter methods (follow-up PR)

Related PRs

• Depends on: [serve] [taskiq - 1/n] Enable async task handler support in task_handler decorator #60977 ([taskiq - 1/n] Enable async task handlers) — merged
• Follow-up: [taskiq - 3/n] Full adapter implementation + E2E tests
• Follow-up: [taskiq - 4/n] Retry, DLQ middleware and reliability tests

🤖 Generated with Claude Code

[gemini-code-assist]

Warning

You have reached your daily quota limit. Please wait up to 24 hours and I will start processing your requests again!

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.