# Connectors (/configure/reference/connectors)
KubeMQ ships ten connectors — the **MCP** and **A2A (agents)** agent platforms,
**CloudEvents**, and seven wire-protocol connectors: **MQTT**, **AMQP 0.9.1**, **AMQP 1.0**,
**STOMP**, **Kafka**, **AWS** (SQS/SNS), and **GCP Pub/Sub**. The three HTTP-server connectors
(MCP, A2A, CloudEvents) are **on by default**; the **seven wire-protocol connectors are opt-in
(disabled by default)** — each opens a new network port and must be explicitly enabled.
Each setting is shown for both targets — Docker single-node (`config.yaml` key · env var)
and Kubernetes/Helm (`spec.*` path). A dash (`—`) in the Helm/CRD column means the setting
is not available on that surface (it is `config.yaml`/env-var-only — supply it through a
mounted config file or a raw pod env var, never a typed CRD field).
Connector environment-variable prefixes follow the acronym rule: all-caps acronym
segments **drop** the underscore (`CONNECTORSMCP_*`, `CONNECTORSCE_*`, `CONNECTORSMQTT_*`,
`CONNECTORSA2_A_*`), while Title-case segments **keep** it (`CONNECTORS_AMQP_*`,
`CONNECTORS_AMQP10_*`, `CONNECTORS_STOMP_*`, `CONNECTORS_KAFKA_*`, `CONNECTORS_AWS_*`,
`CONNECTORS_GCP_*`). See [the reference legend](/configure/reference) for the full rule
and the silently-ignored wrong twin.
## Enabling and disabling a connector [#enabling-and-disabling-a-connector]
The toggle shape differs by connector type and by target:
* **HTTP-server connectors (MCP, A2A, CloudEvents):** always on by default. Docker turns
them **off** with `enable: false`; Kubernetes/Helm turns them off with `disabled: true`
(omit the key while the connector is on).
* **Wire-protocol connectors (MQTT, AMQP 0.9.1, AMQP 1.0, STOMP, Kafka, AWS, GCP Pub/Sub):**
**disabled by default** (opt-in). Docker turns them **on** with `enable: true`; Kubernetes/Helm
turns them on with `enabled: true` (a positive-sense field — omitting it leaves the connector
off).
The snippets below show turning CloudEvents off (HTTP-server connector, uses `disabled:`) and
enabling MQTT (wire-protocol connector, uses `enabled:`). See the
[Docker guide](/configure/docker) and the
[Kubernetes guide](/configure/kubernetes) for complete, runnable configurations.
```yaml title="config.yaml"
connectors:
ce:
enable: false
```
```yaml title="values.yaml"
ce:
disabled: true
```
```bash title="docker run"
docker run -e CONNECTORSMQTT_ENABLE=true ...
```
```yaml title="values.yaml"
mqtt:
enabled: true
```
## Service exposure & session affinity [#service-exposure--session-affinity]
Every wire connector carries the same Kubernetes exposure surface on the CRD, alongside
its protocol settings. These are Kubernetes-only — on Docker you publish a port with
`docker run -p`.
| Field | Type | Default | Valid values | Helm/CRD path | Notes |
| ------------------- | ------------- | ----------- | ----------------------------------------- | ---------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------- |
| Service exposure | string (enum) | `ClusterIP` | `ClusterIP` / `NodePort` / `LoadBalancer` | `spec..expose` | Type of the connector's `Service`. |
| Session affinity | string (enum) | `None` | `None` / `ClientIP` | `spec..sessionAffinity` | Pins a client to one replica. **Required for AWS and GCP** — see the callout below. |
| Node port | int32 | *unset* | 30000–32767 | `spec..nodePort` | Honored **only** when `expose: NodePort`. Unset ⇒ the kernel assigns one, which you cannot configure into a client ahead of the install. |
| TLS node port | int32 | *unset* | 30000–32767 | `spec..tlsNodePort` | Same, for the connector's TLS listener. |
| WebSocket node port | int32 | *unset* | 30000–32767 | `spec.mqtt.wsNodePort` | MQTT only — its WebSocket listener. |
Which connector has which:
| Connector | `expose` | `sessionAffinity` | `nodePort` | `tlsNodePort` | `wsNodePort` |
| ----------- | -------------- | ----------------- | ---------- | ------------- | ------------ |
| MQTT | ✅ | ✅ | ✅ | ✅ | ✅ |
| AMQP 0.9.1 | ✅ | ✅ | ✅ | ✅ | — |
| AMQP 1.0 | ✅ | ✅ | ✅ | ✅ | — |
| STOMP | ✅ | ✅ | ✅ | ✅ | — |
| Kafka | ✅ | ✅ | ✅ | ✅ | — |
| AWS | ✅ | ✅ | ✅ | — | — |
| GCP Pub/Sub | ✅ | ✅ | ✅ | — | — |
| CloudEvents | — (rides REST) | — | — | — | — |
**Session affinity is not optional for the AWS and GCP connectors on a multi-replica
cluster.** Both protocols hand the client a token that only the replica that minted it can
honor:
* **AWS** — an SQS **receipt handle** is bound to the replica that issued it, and it is the
only way to delete a message. A delete that lands on another replica is refused, the
message reappears at the visibility timeout, **and the queue never drains**. A client
holding a single keep-alive connection is pinned by accident and never sees this; it
bites when the connection breaks, or when the client does not pool connections.
* **GCP Pub/Sub** — an **ack id** minted by one replica and acked on another is refused,
with no disruption required to trigger it. Ordered subscriptions additionally need
stickiness to preserve per-key ordering.
Set `sessionAffinity: ClientIP` on both:
```yaml title="values.yaml"
aws:
enabled: true
expose: NodePort
sessionAffinity: ClientIP
nodePort: 30171
gcp:
enabled: true
expose: NodePort
sessionAffinity: ClientIP
nodePort: 32439
```
**`ClientIP` affinity is unreliable when clients share a NAT or egress IP** — every client
behind it looks like one address and lands on one replica. Where an ingress exists, prefer
**cookie-based affinity on the ingress** instead.
**AMQP 0.9.1 and AMQP 1.0 share one listener, and one Service.** Both ride `5672`/`5671`;
enabling both is supported and both are served through the **`-amqp`** Service.
Enabling `amqp10` additionally creates a **`-amqp10`** Service as a
discoverability alias onto that same listener. `expose` and `sessionAffinity` are
**shared** between the two blocks — last one wins — so set them consistently. The operator
raises an `AmqpSessionAffinityConflict` warning event if the two disagree.
**Kafka exposure on a multi-replica cluster needs one address per broker.** Kafka hands
every client an address **per broker** in its Metadata response, so a single Service —
`ClusterIP`, `NodePort`, or `LoadBalancer` alike — is one address that round-robins across
all replicas and cannot address a 3-broker cluster. Setting `kafka.expose: LoadBalancer`
on a multi-replica cluster does **not** give you working external Kafka.
* **In-cluster, any replica count:** leave `spec.kafka.advertisedHost` unset. The operator
derives per-broker addresses from the pods' stable DNS names and gives each pod its own
advertised host. Nothing to configure.
* **External, single replica:** `expose` + `advertisedHost` works.
* **External, multi-replica:** you must provision one client-reachable address per broker
— a Service or LoadBalancer per pod, or a per-pod NodePort — and list them in
`spec.kafka.peers`. The operator does not create per-broker addressing for you. Leave
`advertisedHost` unset in this case: with a peer map each broker advertises itself from
its own entry, and one `advertisedHost` could only ever be right for one of them.
**CloudEvents has no port or Service of its own.** It rides the server's shared HTTP
listener alongside REST, and is reached and exposed through **`spec.rest.expose` /
`spec.rest.nodePort`**. There is deliberately no `spec.ce.expose`.
## MCP [#mcp]
The Model Context Protocol agent platform, served on the shared HTTP server. Env prefix
`CONNECTORSMCP_*` (all-caps `MCP` collapses the underscore after `CONNECTORS`); CRD group
`spec.mcp.*`. This is an **HTTP-family connector — on by default** (opt-out): Docker turns it
off with `enable: false`, Kubernetes/Helm with `spec.mcp.disabled: true`.
| Setting | Type | Default | Valid values | Docker (config.yaml key · env var) | Helm/CRD path | Notes |
| ---------------- | --------- | ----------- | ---------------------- | -------------------------------------------------------------------------- | ----------------------------- | ------------------------------------------------------------------------------------------ |
| Enable / disable | bool | `true` (on) | true / false | `connectors.mcp.enable` · `CONNECTORSMCP_ENABLE` | `spec.mcp.disabled` | Inverted boolean: Docker `enable: false` turns it off; Helm `disabled: true` turns it off. |
| Tool timeout (s) | int | `300` | `> 0` | `connectors.mcp.tooltimeoutseconds` · `CONNECTORSMCP_TOOL_TIMEOUT_SECONDS` | `spec.mcp.toolTimeoutSeconds` | Must be positive (rejected if ≤ 0). CRD minimum 1. |
| Trusted origins | string\[] | `["auto"]` | origin list / `"auto"` | `connectors.mcp.trustedorigins` · `CONNECTORSMCP_TRUSTED_ORIGINS` | `spec.mcp.trustedOrigins` | `auto` derives allowed origins from the request host. |
**The MCP env prefix is `CONNECTORSMCP_` — no underscore between `CONNECTORS` and `MCP`.** The
natural `CONNECTORS_MCP_*` form does **not** bind — the server starts, accepts the variable
without error, and silently ignores it. Unlike CloudEvents, MCP has no natural-name alias, so
`CONNECTORSMCP_ENABLE` / `CONNECTORSMCP_TOOL_TIMEOUT_SECONDS` / `CONNECTORSMCP_TRUSTED_ORIGINS`
are the only working names.
## A2A (Agents) [#a2a-agents]
The agent-to-agent platform, served on the shared HTTP server. Env prefix `CONNECTORSA2_A_*`;
the CRD group is `spec.agents.*` (note the group name differs from the connector name). This is
an **HTTP-family connector — on by default** (opt-out): Docker turns it off with `enable: false`,
Kubernetes/Helm with `spec.agents.disabled: true`.
| Setting | Type | Default | Valid values | Docker (config.yaml key · env var) | Helm/CRD path | Notes |
| -------------------------- | --------- | ------------------ | ------------------------ | ---------------------------------------------------------------------------------- | ----------------------------------- | ----------------------------------------------------------------------------------------------------------------------------- |
| Enable / disable | bool | `true` (on) | true / false | `connectors.a2a.enable` · `CONNECTORSA2_A_ENABLE` | `spec.agents.disabled` | Inverted boolean. |
| Agent TTL (s) | int | `300` | `> 0` | `connectors.a2a.agentttlseconds` · `CONNECTORSA2_A_AGENT_TTL_SECONDS` | `spec.agents.agentTtlSeconds` | Must be positive. CRD json tag is `agentTtlSeconds` (lowercase `tl`). |
| Default timeout (s) | int | `300` | `> 0` | `connectors.a2a.defaulttimeoutseconds` · `CONNECTORSA2_A_DEFAULT_TIMEOUT_SECONDS` | `spec.agents.defaultTimeoutSeconds` | Must be positive. |
| Max timeout (s) | int | `3600` | `≥ default timeout` | `connectors.a2a.maxtimeoutseconds` · `CONNECTORSA2_A_MAX_TIMEOUT_SECONDS` | `spec.agents.maxTimeoutSeconds` | Cross-field: must be ≥ Default timeout (enforced server-side, not by the CRD schema). |
| Max agents | int | `0` | `≥ 0` (`0` = unlimited) | `connectors.a2a.maxagents` · `CONNECTORSA2_A_MAX_AGENTS` | `spec.agents.maxAgents` | `0` = unlimited. |
| Max SSE idle (s) | int | `300` | `> 0` | `connectors.a2a.maxsseidleseconds` · `CONNECTORSA2_A_MAX_SSE_IDLE_SECONDS` | `spec.agents.maxSseIdleSeconds` | Must be positive. CRD json tag is `maxSseIdleSeconds`. |
| Trusted origins | string\[] | `["auto"]` | origin list / `"auto"` | `connectors.a2a.trustedorigins` · `CONNECTORSA2_A_TRUSTED_ORIGINS` | `spec.agents.trustedOrigins` | |
| Agent max response (bytes) | int64 | `10485760` (10 MB) | `≥ 0` | `connectors.a2a.agentmaxresponsebytes` · `CONNECTORSA2_A_AGENT_MAX_RESPONSE_BYTES` | `spec.agents.agentMaxResponseBytes` | Caps a downstream agent's response body. |
| Agent TLS skip verify | bool | `false` | true / false | `connectors.a2a.agenttlsskipverify` · `CONNECTORSA2_A_AGENT_TLS_SKIP_VERIFY` | `spec.agents.agentTlsSkipVerify` | CRD json tag is `agentTlsSkipVerify`. |
| Agent max concurrency | int | `100` | any int (`≤ 0` accepted) | `connectors.a2a.agentmaxconcurrency` · `CONNECTORSA2_A_AGENT_MAX_CONCURRENCY` | `spec.agents.agentMaxConcurrency` | `≤ 0` is **accepted** and silently clamped back to `100` — it does not error, and there is no output saying it was rewritten. |
| Metrics retention (h) | int | `168` (7 days) | `> 0` | `connectors.a2a.metricsretentionhours` · `CONNECTORSA2_A_METRICS_RETENTION_HOURS` | `spec.agents.metricsRetentionHours` | Must be positive. Retention window for per-agent metrics. |
**The A2A env prefix is `CONNECTORSA2_A_` — not `CONNECTORS_A2A_`.** The `convertEnvFormat`
rule splits `A2A` into `A2_A` (the regex breaks between the digit and the trailing `A`), so the
agent variables read `CONNECTORSA2_A_MAX_AGENTS`, `CONNECTORSA2_A_AGENT_TTL_SECONDS`, and so on.
Neither `CONNECTORS_A2A_*` nor `CONNECTORSA2A_*` binds — both are silently ignored. Note also
that the CRD group is `spec.agents.*`, not `spec.a2a.*`.
## CloudEvents [#cloudevents]
The CloudEvents connector, served on the shared HTTP server. Env prefix `CONNECTORSCE_*`;
CRD group `spec.ce.*`. This is an **HTTP-family connector — on by default** (opt-out): Docker
turns it off with `enable: false`, Kubernetes/Helm with `spec.ce.disabled: true`.
| Setting | Type | Default | Valid values | Docker (config.yaml key · env var) | Helm/CRD path | Notes |
| ------------------- | ---- | ----------- | ----------------------- | ----------------------------------------------------------------------- | --------------------------- | ------------------------------------------------------ |
| Enable / disable | bool | `true` (on) | true / false | `connectors.ce.enable` · `CONNECTORSCE_ENABLE` | `spec.ce.disabled` | Inverted boolean. |
| Timeout (s) | int | `60` | `> 0` | `connectors.ce.timeoutseconds` · `CONNECTORSCE_TIMEOUT_SECONDS` | `spec.ce.timeoutSeconds` | Must be positive. |
| Sub buffer size | int | `100` | `1`–`10000` | `connectors.ce.subbuffsize` · `CONNECTORSCE_SUB_BUFF_SIZE` | `spec.ce.subBuffSize` | Rejected if ≤ 0 or > 10000. |
| Max SSE idle (s) | int | `300` | `> 0` | `connectors.ce.maxsseidleseconds` · `CONNECTORSCE_MAX_SSE_IDLE_SECONDS` | `spec.ce.maxSseIdleSeconds` | Must be positive. CRD json tag is `maxSseIdleSeconds`. |
| Max SSE connections | int | `0` | `≥ 0` (`0` = unlimited) | `connectors.ce.maxsseconnections` · `CONNECTORSCE_MAX_SSE_CONNECTIONS` | `spec.ce.maxSseConnections` | `0` = unlimited. CRD json tag is `maxSseConnections`. |
**CloudEvents accepts both env forms.** The primary name is the collapsed `CONNECTORSCE_*`
(e.g. `CONNECTORSCE_ENABLE`), but CE is the one connector that also binds the natural
`CONNECTORS_CE_*` alias (`CONNECTORS_CE_ENABLE`, `CONNECTORS_CE_TIMEOUT_SECONDS`, …). Both
resolve to the same setting — this compensating alias exists only for CloudEvents; MCP, A2A,
and MQTT do **not** have it.
**But the server warns that the alias is IGNORED, and that warning is wrong.** The
unknown-variable warner does not track the alias binding, so a `CONNECTORS_CE_*` variable is
reported as ignored while its value is being applied. Meanwhile a typo in the *collapsed*
form (`CONNECTORSC_ENABLE`) produces no warning at all, because collapsed names sit outside
the warner's namespace list. See [the reference legend](/configure/reference) for the
full picture.
## MQTT [#mqtt]
The MQTT 3.1.1 / 5.0 wire protocol. Env prefix `CONNECTORSMQTT_*` (no underscore after
`CONNECTORS`); CRD group `spec.mqtt.*`. This is a **wire-protocol connector — opt-in (disabled
by default)**: Docker turns it on with `enable: true`, Kubernetes/Helm with `spec.mqtt.enabled:
true` (a positive-sense `*bool` — omitting it leaves MQTT off). Ports are `string` server-side
(`""` disables a listener) and `int32` on the CRD.
| Setting | Type | Default | Valid values | Docker (config.yaml key · env var) | Helm/CRD path | Notes |
| ------------------------------------------- | ------ | -------------------- | --------------------------- | --------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------- | ------------------------------------------------------------------------------------------------------------ |
| Enable / disable | bool | **`false` (opt-in)** | true / false | `connectors.mqtt.enable` · `CONNECTORSMQTT_ENABLE` | `spec.mqtt.enabled` | Opt-in: set `true` to open ports 1883/8883/8083. Helm uses positive-sense `enabled: true`. |
| Port | string | `1883` | port / `""` | `connectors.mqtt.port` · `CONNECTORSMQTT_PORT` | `spec.mqtt.port` | Plaintext TCP listener; `""` disables it. CRD type int32 (1–65535). |
| TLS port | string | `8883` | port / `""` | `connectors.mqtt.tlsport` · `CONNECTORSMQTT_TLS_PORT` | `spec.mqtt.tlsPort` | TLS listener; **active only when Security mode ≠ None**. `""` disables. CRD int32 (1–65535). |
| WebSocket port | string | `8083` | port / `""` | `connectors.mqtt.wsport` · `CONNECTORSMQTT_WS_PORT` | `spec.mqtt.wsPort` | WebSocket listener; `""` disables. CRD int32 (1–65535). |
| Default pattern | enum | `events` | `events` / `store` / `none` | `connectors.mqtt.defaultpattern` · `CONNECTORSMQTT_DEFAULT_PATTERN` | `spec.mqtt.defaultPattern` | KubeMQ pattern for prefixless topics. |
| Sub buffer size | int | `100` | `1`–`10000` | `connectors.mqtt.subbuffsize` · `CONNECTORSMQTT_SUB_BUFF_SIZE` | `spec.mqtt.subBuffSize` | Rejected if ≤ 0 or > 10000. |
| Queue ACK timeout (s) | int | `30` | `> 0` | `connectors.mqtt.queueacktimeoutseconds` · `CONNECTORSMQTT_QUEUE_ACK_TIMEOUT_SECONDS` | `spec.mqtt.queueAckTimeoutSeconds` | Must be positive. |
| RPC timeout (s) | int | `30` | `> 0` | `connectors.mqtt.rpctimeoutseconds` · `CONNECTORSMQTT_RPC_TIMEOUT_SECONDS` | `spec.mqtt.rpcTimeoutSeconds` | Must be positive. |
| RPC max pending | int | `1024` | `> 0` | `connectors.mqtt.rpcmaxpending` · `CONNECTORSMQTT_RPC_MAX_PENDING` | `spec.mqtt.rpcMaxPending` | Must be positive. |
| Detail history enabled | bool | `true` | true / false | `connectors.mqtt.detailhistoryenabled` · `CONNECTORSMQTT_DETAIL_HISTORY_ENABLED` | `spec.mqtt.detailHistoryEnabled` | Master switch for per-entity (client/subscription) detail-page history recording. |
| Detail history max entities | int | `5000` | `≥ 0` (`0` = unbounded) | `connectors.mqtt.detailhistorymaxentities` · `CONNECTORSMQTT_DETAIL_HISTORY_MAX_ENTITIES` | `spec.mqtt.detailHistoryMaxEntities` | Caps tracked per-entity history keys; new keys refused beyond it. Validated even when history is off. |
| Capabilities · max clients | int64 | `0` | `≥ 0` (`0` = unlimited) | `connectors.mqtt.capabilities.maxclients` · `CONNECTORSMQTT_CAPABILITIES_MAX_CLIENTS` | `spec.mqtt.capabilities.maxClients` | `0` = unlimited (explicit opt-in; startup logs a WARN). Never clamped. |
| Capabilities · max packet size (bytes) | uint32 | `4194304` (4 MB) | `0` or `1`–`4294967295` | `connectors.mqtt.capabilities.maxpacketsizebytes` · `CONNECTORSMQTT_CAPABILITIES_MAX_PACKET_SIZE_BYTES` | `spec.mqtt.capabilities.maxPacketSizeBytes` | A `0` is clamped back to 4 MB (0 = "unlimited" is a DoS footgun); effective value logged at startup. |
| Capabilities · receive maximum | uint16 | `1024` | `0` or `1`–`65535` | `connectors.mqtt.capabilities.receivemaximum` · `CONNECTORSMQTT_CAPABILITIES_RECEIVE_MAXIMUM` | `spec.mqtt.capabilities.receiveMaximum` | A `0` is clamped back to 1024. |
| Capabilities · max inflight | uint16 | `8192` | `0`–`65535` | `connectors.mqtt.capabilities.maxinflight` · `CONNECTORSMQTT_CAPABILITIES_MAX_INFLIGHT` | `spec.mqtt.capabilities.maxInflight` | Not clamped. |
| Capabilities · max session expiry (s) | uint32 | `3600` | `0` or `1`–`4294967295` | `connectors.mqtt.capabilities.maxsessionexpiryseconds` · `CONNECTORSMQTT_CAPABILITIES_MAX_SESSION_EXPIRY_SECONDS` | `spec.mqtt.capabilities.maxSessionExpirySeconds` | A `0` is clamped back to 3600. |
| Capabilities · max message expiry (s) | int64 | `86400` | `≥ 0` | `connectors.mqtt.capabilities.maxmessageexpiryseconds` · `CONNECTORSMQTT_CAPABILITIES_MAX_MESSAGE_EXPIRY_SECONDS` | `spec.mqtt.capabilities.maxMessageExpirySeconds` | Rejected if negative. |
| Capabilities · max QoS | byte | `2` | `0`–`2` | `connectors.mqtt.capabilities.maxqos` · `CONNECTORSMQTT_CAPABILITIES_MAX_QOS` | `spec.mqtt.capabilities.maxQos` | Rejected if > 2. |
| Capabilities · min protocol version | byte | `4` | `4` / `5` | `connectors.mqtt.capabilities.minprotocolversion` · `CONNECTORSMQTT_CAPABILITIES_MIN_PROTOCOL_VERSION` | `spec.mqtt.capabilities.minProtocolVersion` | `4` = MQTT 3.1.1, `5` = MQTT 5.0. |
| Capabilities · max subscriptions per client | int | `1000` | `≥ 0` (`0` = unlimited) | `connectors.mqtt.capabilities.maxsubscriptionsperclient` · `CONNECTORSMQTT_CAPABILITIES_MAX_SUBSCRIPTIONS_PER_CLIENT` | `spec.mqtt.capabilities.maxSubscriptionsPerClient` | Caps distinct subscription filters per client; excess SUBSCRIBEs get SUBACK 0x97 (DoS cap). `0` = unlimited. |
**The MQTT env prefix is `CONNECTORSMQTT_` — no underscore between `CONNECTORS` and `MQTT`.**
The natural `CONNECTORS_MQTT_*` form does **not** bind and is silently ignored. This applies to
every MQTT variable, including the nested `CONNECTORSMQTT_CAPABILITIES_*` keys. Unlike
CloudEvents, MQTT has no natural-name alias.
**Some MQTT capabilities are forced, not configurable.** At server construction the connector
pins `RetainAvailable = 0` (retained messages rejected), `SharedSubAvailable = 1`, and
`WildcardSubAvailable = 1` regardless of config. A partially-specified `capabilities` block
leaves unset safety caps (`maxPacketSizeBytes`, `receiveMaximum`, `maxSessionExpirySeconds`) at
`0`, which the server clamps back to their safe defaults rather than treating as "unlimited" —
the effective caps are logged at startup. At least one listener port (`port`, `tlsPort`, or
`wsPort`) must be non-empty, and no two may share the same value.
## AMQP 0.9.1 [#amqp-091]
The AMQP 0.9.1 / RabbitMQ wire protocol. Env prefix `CONNECTORS_AMQP_*`; CRD group
`spec.amqp.*`.
| Setting | Type | Default | Valid values | Docker (config.yaml key · env var) | Helm/CRD path | Notes |
| --------------------- | ------ | -------------------- | --------------------------------------------------- | ---------------------------------------------------------------------------- | ----------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| Enable | bool | **`false` (opt-in)** | true / false | `connectors.amqp.enable` · `CONNECTORS_AMQP_ENABLE` | `spec.amqp.enabled` | Opt-in wire connector: Docker `enable: true`, Helm `enabled: true` (positive-sense). Opens ports 5672/5671 (shared mux with AMQP 1.0). |
| Port | int | `5672` | `0`–`65535` (`0` disables) | `connectors.amqp.port` · `CONNECTORS_AMQP_PORT` | `spec.amqp.port` | Plaintext listener, shared with AMQP 1.0. `Port` and `TlsPort` cannot both be `0` when enabled. CRD accepts `1`–`65535`. |
| TLS port | int | `5671` | `0`–`65535` (`0` disables) | `connectors.amqp.tlsport` · `CONNECTORS_AMQP_TLS_PORT` | `spec.amqp.tlsPort` | TLS listener. |
| Heartbeat (s) | int | `60` | ≥ `0` | `connectors.amqp.heartbeatseconds` · `CONNECTORS_AMQP_HEARTBEAT_SECONDS` | `spec.amqp.heartbeatSeconds` | |
| Frame max (bytes) | int | `131072` | ≥ `4096` (accepted; > 512 MiB is clamped down) | `connectors.amqp.framemax` · `CONNECTORS_AMQP_FRAME_MAX` | `spec.amqp.frameMax` | Silently **clamped down to 536870912 (512 MiB)** if set higher (a stderr warning is emitted); a value ≥ 2³² would otherwise narrow to `0` and disable the codec frame-size cap. |
| Channel max | int | `2047` | `1`–`65535` | `connectors.amqp.channelmax` · `CONNECTORS_AMQP_CHANNEL_MAX` | `spec.amqp.channelMax` | |
| Max connections | int | `1000` | ≥ `0` (`0` = unlimited) | `connectors.amqp.maxconnections` · `CONNECTORS_AMQP_MAX_CONNECTIONS` | `spec.amqp.maxConnections` | `0` = unlimited. |
| Max body size (bytes) | int | `104857600` | > `0` | `connectors.amqp.maxbodysize` · `CONNECTORS_AMQP_MAX_BODY_SIZE` | `spec.amqp.maxBodySize` | |
| Default vhost | string | `default` | non-empty; no whitespace or `;:*>`; no trailing `.` | `connectors.amqp.defaultvhost` · `CONNECTORS_AMQP_DEFAULT_VHOST` | `spec.amqp.defaultVhost` | Becomes a channel segment, so it must pass the channel-charset rules. |
| Get batch size | int | `32` | `1`–`1024` | `connectors.amqp.getbatchsize` · `CONNECTORS_AMQP_GET_BATCH_SIZE` | `spec.amqp.getBatchSize` | Must also be ≤ `queue.maxNumberOfMessages` (cross-checked in top-level config validation). |
| Dead-letter max hops | int | `16` | ≥ `1` | `connectors.amqp.deadlettermaxhops` · `CONNECTORS_AMQP_DEAD_LETTER_MAX_HOPS` | `spec.amqp.deadLetterMaxHops` | |
| Max receive count | int | `0` | ≥ `0` (`0` = unlimited) | `connectors.amqp.maxreceivecount` · `CONNECTORS_AMQP_MAX_RECEIVE_COUNT` | `spec.amqp.maxReceiveCount` | `0` = unlimited. Must also be ≤ `queue.maxReceiveCount` (cross-checked in top-level config validation). |
**Two cross-checks against the queue limits.** The server rejects the config unless
`amqp.maxReceiveCount ≤ queue.maxReceiveCount` and `amqp.getBatchSize ≤
queue.maxNumberOfMessages`. Keep the AMQP values within the queue ceilings (see
[Storage & Queues](/configure/reference/storage-queues)).
## AMQP 1.0 [#amqp-10]
The AMQP 1.0 wire protocol (also the JMS / Qpid path). Env prefix `CONNECTORS_AMQP10_*`;
CRD group `spec.amqp10.*`.
| Setting | Type | Default | Valid values | Docker (config.yaml key · env var) | Helm/CRD path | Notes |
| ------------------------ | ----- | -------------------- | ------------------------------------------------------------- | ---------------------------------------------------------------------------------------------- | -------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------ |
| Enable | bool | **`false` (opt-in)** | true / false | `connectors.amqp10.enable` · `CONNECTORS_AMQP10_ENABLE` | `spec.amqp10.enabled` | Opt-in wire connector. Opens ports 5672/5671 (shared mux with AMQP 0.9.1). |
| Port | int | `5672` | `0`–`65535` (`0` disables) | `connectors.amqp10.port` · `CONNECTORS_AMQP10_PORT` | `spec.amqp10.port` | Shared with AMQP 0.9.1 — a `Port == Amqp.Port` collision is intentionally allowed (the mux dedupes the bind). `Port` and `TlsPort` cannot both be `0`. |
| TLS port | int | `5671` | `0`–`65535` (`0` disables) | `connectors.amqp10.tlsport` · `CONNECTORS_AMQP10_TLS_PORT` | `spec.amqp10.tlsPort` | |
| Max frame size (bytes) | int | `131072` | ≥ `512` | `connectors.amqp10.maxframesize` · `CONNECTORS_AMQP10_MAX_FRAME_SIZE` | `spec.amqp10.maxFrameSize` | Spec floor 512. |
| Max message size (bytes) | int64 | `104857600` | > `0` | `connectors.amqp10.maxmessagesize` · `CONNECTORS_AMQP10_MAX_MESSAGE_SIZE` | `spec.amqp10.maxMessageSize` | `int64`; matches the AMQP 0.9.1 `maxBodySize` default (100 MB). |
| Session max | int | `256` | `1`–`65535` | `connectors.amqp10.sessionmax` · `CONNECTORS_AMQP10_SESSION_MAX` | `spec.amqp10.sessionMax` | |
| Max links per session | int | `256` | ≥ `1` | `connectors.amqp10.maxlinkspersession` · `CONNECTORS_AMQP10_MAX_LINKS_PER_SESSION` | `spec.amqp10.maxLinksPerSession` | |
| Max connections | int | `1000` | ≥ `0` (`0` = unlimited) | `connectors.amqp10.maxconnections` · `CONNECTORS_AMQP10_MAX_CONNECTIONS` | `spec.amqp10.maxConnections` | `0` = unlimited. |
| Idle timeout (s) | int | `120` | ≥ `0` (`0` = disabled) | `connectors.amqp10.idletimeoutseconds` · `CONNECTORS_AMQP10_IDLE_TIMEOUT_SECONDS` | `spec.amqp10.idleTimeoutSeconds` | `0` = disabled. |
| Default pattern | enum | `queues` | `queues` / `events` / `events-store` / `commands` / `queries` | `connectors.amqp10.defaultpattern` · `CONNECTORS_AMQP10_DEFAULT_PATTERN` | `spec.amqp10.defaultPattern` | KubeMQ pattern mapped from AMQP addresses. |
| Get batch size | int | `32` | `1`–`1024` | `connectors.amqp10.getbatchsize` · `CONNECTORS_AMQP10_GET_BATCH_SIZE` | `spec.amqp10.getBatchSize` | |
| Max unsettled per link | int | `1024` | ≥ `1` | `connectors.amqp10.maxunsettledperlink` · `CONNECTORS_AMQP10_MAX_UNSETTLED_PER_LINK` | `spec.amqp10.maxUnsettledPerLink` | |
| Default RPC timeout (s) | int | `30` | ≥ `1` | `connectors.amqp10.defaultrpctimeoutseconds` · `CONNECTORS_AMQP10_DEFAULT_RPC_TIMEOUT_SECONDS` | `spec.amqp10.defaultRpcTimeoutSeconds` | |
| RPC max pending | int | `512` | ≥ `1` | `connectors.amqp10.rpcmaxpending` · `CONNECTORS_AMQP10_RPC_MAX_PENDING` | `spec.amqp10.rpcMaxPending` | |
**AMQP 1.0 shares port `5672` (and TLS `5671`) with AMQP 0.9.1.** The two protocols are
demultiplexed on a single shared listener, so running both on the default ports is fine.
If you set a **different** port for one, keep the pair consistent so clients reach the
listener you intend.
## STOMP [#stomp]
The STOMP 1.0 / 1.1 / 1.2 wire protocol. Env prefix `CONNECTORS_STOMP_*`; CRD group
`spec.stomp.*`.
| Setting | Type | Default | Valid values | Docker (config.yaml key · env var) | Helm/CRD path | Notes |
| --------------------- | ------ | -------------------- | --------------------------------------------- | ---------------------------------------------------------------------------------------- | ----------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------- |
| Enable | bool | **`false` (opt-in)** | true / false | `connectors.stomp.enable` · `CONNECTORS_STOMP_ENABLE` | `spec.stomp.enabled` | Opt-in wire connector. Opens ports 61613/61614. |
| Port | string | `61613` | `""` (disabled) or `1`–`65535` | `connectors.stomp.port` · `CONNECTORS_STOMP_PORT` | `spec.stomp.port` | Plaintext listener. String on the server (`""` disables); the CRD takes an `int32` (`1`–`65535`). `Port` must differ from `TlsPort`. |
| TLS port | string | `61614` | `""` (disabled) or `1`–`65535` | `connectors.stomp.tlsport` · `CONNECTORS_STOMP_TLS_PORT` | `spec.stomp.tlsPort` | TLS listener. At least one of `Port`/`TlsPort` must be set. |
| Default pattern | enum | `events` | `events` / `queues` / `store` / `none` | `connectors.stomp.defaultpattern` · `CONNECTORS_STOMP_DEFAULT_PATTERN` | `spec.stomp.defaultPattern` | KubeMQ pattern for bare (prefixless) destinations. |
| Sub buffer size | int | `100` | `1`–`10000` | `connectors.stomp.subbuffsize` · `CONNECTORS_STOMP_SUB_BUFF_SIZE` | `spec.stomp.subBuffSize` | Events deliver-channel buffer. |
| Max connections | int | `1000` | ≥ `0` (`0` = unlimited) | `connectors.stomp.maxconnections` · `CONNECTORS_STOMP_MAX_CONNECTIONS` | `spec.stomp.maxConnections` | `0` = unlimited. |
| Max body size (bytes) | int | `104857600` | > `0` | `connectors.stomp.maxbodysize` · `CONNECTORS_STOMP_MAX_BODY_SIZE` | `spec.stomp.maxBodySize` | |
| Heartbeat (ms) | int | `10000` | ≥ `0` (`0` = disabled) | `connectors.stomp.heartbeatms` · `CONNECTORS_STOMP_HEARTBEAT_MS` | `spec.stomp.heartbeatMs` | Advertised sx,sy; `0` disables the server side. |
| Queue ACK timeout (s) | int | `30` | > `0` | `connectors.stomp.queueacktimeoutseconds` · `CONNECTORS_STOMP_QUEUE_ACK_TIMEOUT_SECONDS` | `spec.stomp.queueAckTimeoutSeconds` | |
| RPC timeout (s) | int | `30` | > `0` (accepted; > `2147483` is clamped down) | `connectors.stomp.rpctimeoutseconds` · `CONNECTORS_STOMP_RPC_TIMEOUT_SECONDS` | `spec.stomp.rpcTimeoutSeconds` | Silently **clamped down to `2147483`** (\~24.8 days) if set higher, so `timeout × 1000` cannot overflow the `int32` RPC-bridge deadline. |
| RPC max pending | int | `1024` | > `0` | `connectors.stomp.rpcmaxpending` · `CONNECTORS_STOMP_RPC_MAX_PENDING` | `spec.stomp.rpcMaxPending` | In-flight RPC cap. |
## Kafka [#kafka]
The embedded **Kafka drop-in connector** — KubeMQ speaks the native Kafka wire protocol,
so real `librdkafka`/`kcat`/Java clients connect unchanged. Env prefix `CONNECTORS_KAFKA_*`;
CRD group `spec.kafka.*`. As of v3.1 the connector is compiled into the **default build**
(no build tag) and gated purely at runtime by `Enable`.
Eight fields are exposed as typed CRD fields (`spec.kafka.*`) — including the Service-exposure
type; the remaining advanced knobs — fourteen scalars **plus the six-field OAUTHBEARER
block** — are **`config.yaml`/env-var-only** (Helm/CRD path `—`) and were deliberately
deferred from the CRD in v3.1 — set them via a mounted config file or a raw pod env var. The
SASL credential store is **secret/file-only** (no env var, no CRD field).
See the [Kafka connector overview](/connectors/kafka) and the
[migration guide](/connectors/kafka/how-to/migrate-from-kafka) for adoption
planning, fitness assessment, and cutover tooling beyond this config reference.
**Kafka requires the `next` storage engine — but it's zero-config.** On a **fresh** store
with Kafka enabled and `Store.Engine` unset, the server **auto-selects `next`** (a `NOTICE`
is logged) — no manual `store.engine=next` step needed. It fails closed only when the store
directory already holds `legacy` data (a config error naming the conflicting directory) or
`Store.Engine=legacy` is set explicitly alongside Kafka. Pinning `STORE_ENGINE=next` skips
the probe entirely and always wins. See
[Storage Engines](/configure/reference/storage-engines#zero-config-engine-selection).
### Core (CRD-exposed) settings [#core-crd-exposed-settings]
| Setting | Type | Default | Valid values | Docker (config.yaml key · env var) | Helm/CRD path | Notes |
| ----------------- | ------------- | -------------------- | ----------------------------------------- | ------------------------------------------------------------------------- | ---------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Enable | bool | **`false` (opt-in)** | true / false | `connectors.kafka.enable` · `CONNECTORS_KAFKA_ENABLE` | `spec.kafka.enabled` | Opt-in wire connector. Opens ports 9092/9093. |
| Port | string | `9092` | `""` (disabled) or `1`–`65535` | `connectors.kafka.port` · `CONNECTORS_KAFKA_PORT` | `spec.kafka.port` | Plaintext listener. **A TLS-only config (`TlsPort` set, `Port` empty) is rejected** — the TLS accept path is not yet wired, so `Port` is currently required. `Port` must differ from `TlsPort`. |
| TLS port | string | `9093` | `""` (disabled) or `1`–`65535` | `connectors.kafka.tlsport` · `CONNECTORS_KAFKA_TLS_PORT` | `spec.kafka.tlsPort` | Reserved TLS listener (see the TLS-only note above). |
| Advertised host | string | `""` | hostname / IP | `connectors.kafka.advertisedhost` · `CONNECTORS_KAFKA_ADVERTISED_HOST` | `spec.kafka.advertisedHost` | The single broker address handed to every client in Metadata/FindCoordinator. **Set it** to the external LoadBalancer DNS / NodePort IP (or the in-cluster Service DNS) — leaving it `""` falls back to the server `Host`, then the pod hostname, which is unreachable off-pod (connect-then-hang). The TLS cert SAN must include this value. |
| Advertised port | int | `0` | `0`–`65535` (`0` = use `Port`) | `connectors.kafka.advertisedport` · `CONNECTORS_KAFKA_ADVERTISED_PORT` | `spec.kafka.advertisedPort` | The external LB/NodePort port. `0` = fall back to `Port` on the config.yaml/env path. **The CRD schema is stricter than the server**: `spec.kafka.advertisedPort` enforces `minimum: 1`, so `0` is rejected on the typed CRD field even though the server itself accepts it. |
| Max connections | int | `1000` | ≥ `0` (`0` = unlimited) | `connectors.kafka.maxconnections` · `CONNECTORS_KAFKA_MAX_CONNECTIONS` | `spec.kafka.maxConnections` | `0` = unlimited. |
| Max message bytes | int | `1048576` | `1`–`1073741824` (1 GiB) | `connectors.kafka.maxmessagebytes` · `CONNECTORS_KAFKA_MAX_MESSAGE_BYTES` | `spec.kafka.maxMessageBytes` | Per-message cap (1 MiB default). Hard ceiling 1 GiB — Kafka frames are `int32`-length-prefixed, so a larger value would truncate. |
| Service exposure | string (enum) | `ClusterIP` | `ClusterIP` / `NodePort` / `LoadBalancer` | — (Docker: `-p` host port mapping) | `spec.kafka.expose` | Kubernetes Service type for the Kafka listener. Kafka also takes `sessionAffinity`, `nodePort`, and `tlsNodePort` — see [Service exposure & session affinity](#service-exposure--session-affinity), including the **multi-replica addressing** limit. |
### Advanced settings (config.yaml / env-var only — no CRD field) [#advanced-settings-configyaml--env-var-only--no-crd-field]
These are on the `configData` allowlist (deferred from the CRD in v3.1). Every one has a
working env var but **no `spec.kafka.*` path** — mount them via config file or a raw pod env var.
| Setting | Type | Default | Valid values | Docker (config.yaml key · env var) | Helm/CRD path | Notes |
| -------------------------------- | --------- | ------------------------------------------------------ | ------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------- | ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| Max fetch wait (ms) | int | `1000` | ≤ `300000` (5 min) | `connectors.kafka.maxfetchwaitms` · `CONNECTORS_KAFKA_MAX_FETCH_WAIT_MS` | — | Ceiling a client's `fetch.max.wait.ms` is clamped to. `≤ 0` falls back to the 1 s default; an over-ceiling value is rejected. |
| Timestamp type | string | `CreateTime` | `""` / `CreateTime` | `connectors.kafka.timestamptype` · `CONNECTORS_KAFKA_TIMESTAMP_TYPE` | — | `LogAppendTime` is **rejected** (deferred); any other value is rejected. |
| Offsets retention (min) | int | `10080` (7 days) | ≤ `52560000` (100 yr) | `connectors.kafka.offsetsretentionminutes` · `CONNECTORS_KAFKA_OFFSETS_RETENTION_MINUTES` | — | Committed-offset expiry. `≤ 0` is floored to the default (no "retain forever"); over-ceiling rejected. |
| Max groups | int | `10000` | ≤ `10000000` | `connectors.kafka.maxgroups` · `CONNECTORS_KAFKA_MAX_GROUPS` | — | Coordinator-wide consumer-group registry cap. `≤ 0` floored to default; over-ceiling rejected. |
| Max topics per request | int | `10000` | ≤ `1000000` | `connectors.kafka.maxtopicsperrequest` · `CONNECTORS_KAFKA_MAX_TOPICS_PER_REQUEST` | — | Per-request distinct-topic cap (DoS fan-out guard). `≤ 0` floored; over-ceiling rejected. |
| Max partitions per request | int | `100000` | ≤ `10000000` | `connectors.kafka.maxpartitionsperrequest` · `CONNECTORS_KAFKA_MAX_PARTITIONS_PER_REQUEST` | — | Per-request distinct-partition cap. `≤ 0` floored; over-ceiling rejected. |
| SCRAM iterations | int | `4096` | ≤ `1000000` | `connectors.kafka.scramiterations` · `CONNECTORS_KAFKA_SCRAM_ITERATIONS` | — | PBKDF2 iteration count for the SCRAM verifier. `≤ 0` floored to 4096 (RFC-7677 minimum); over-ceiling rejected. |
| SASL mechanisms | string\[] | `[]` (offer `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`) | subset of `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`, `OAUTHBEARER` (empty = offer the first three only) | `connectors.kafka.saslmechanisms` · `CONNECTORS_KAFKA_SASL_MECHANISMS` | — | Empty = offer the first three (`PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`). `OAUTHBEARER` is **never** part of the implicit default — it must be listed explicitly, and doing so requires `OAuthBearer.Issuer` set (see OAUTHBEARER authentication below). A non-empty allow-list restricts what `SaslHandshake` offers (e.g. SCRAM-only, dropping cleartext PLAIN); an unknown entry is rejected. |
| Produce byte rate | int | `0` (unlimited) | ≤ `1099511627776` (1 TiB/s) | `connectors.kafka.producebyterate` · `CONNECTORS_KAFKA_PRODUCE_BYTE_RATE` | — | Per-principal produce quota (bytes/s). `0` = unlimited (`ThrottleMillis=0`). `< 0` floored to 0; over-ceiling rejected. |
| Fetch byte rate | int | `0` (unlimited) | ≤ `1099511627776` (1 TiB/s) | `connectors.kafka.fetchbyterate` · `CONNECTORS_KAFKA_FETCH_BYTE_RATE` | — | Fetch-direction twin of the produce quota. |
| Max transaction timeout (ms) | int | `900000` (15 min) | ≤ `86400000` (24 h) | `connectors.kafka.maxtransactiontimeoutms` · `CONNECTORS_KAFKA_MAX_TRANSACTION_TIMEOUT_MS` | — | Ceiling on the `transaction.timeout.ms` a client negotiates. `≤ 0` floored; over-ceiling rejected. |
| Transactional ID expiration (ms) | int64 | `604800000` (7 days) | ≤ `3153600000000` (100 yr) | `connectors.kafka.transactionalidexpirationms` · `CONNECTORS_KAFKA_TRANSACTIONAL_ID_EXPIRATION_MS` | — | Idle-`transactional.id` reaper deadline. `int64`. `≤ 0` floored; over-ceiling rejected. |
| Max transactional IDs | int | `10000` | ≤ `10000000` | `connectors.kafka.maxtransactionalids` · `CONNECTORS_KAFKA_MAX_TRANSACTIONAL_I_DS` | — | Txn-coordinator registry cap. **Env-name trap** (see callout). `≤ 0` floored; over-ceiling rejected. |
| Producer ID block size | int | `1000` | ≤ `1000000` | `connectors.kafka.produceridblocksize` · `CONNECTORS_KAFKA_PRODUCER_ID_BLOCK_SIZE` | — | Producer-ID allocation block. `≤ 0` floored; over-ceiling rejected. |
| Produce pipeline depth | int | `5` | `1`–`5` | `connectors.kafka.producepipelinedepth` · `CONNECTORS_KAFKA_PRODUCE_PIPELINE_DEPTH` | — | Per-partition in-flight produce batches. `5` matches Kafka's idempotent in-flight cap, and is the hard ceiling — above it the broker cannot de-duplicate on replay, so it is rejected. `1` is serial behavior, the emergency rollback. `< 1` is floored to `1` (the safe side), **not** to the default. Worst-case memory is `depth × maxMessageBytes` held per hot partition. |
| Per-broker peer map | string | `""` | `id@host:port,…` | `connectors.kafka.peers` · `CONNECTORS_KAFKA_PEERS` | — | The clustered per-broker **client-reachable** advertised addresses, same grammar as the replication peer map. The id is the peer's raft replica id (= its Kafka broker node id). Meaningful **only** when Kafka is enabled on a clustered `next` cluster — setting it while the connector is on but the cluster is not is **rejected**. A peer's Kafka address must never equal that same id's raft address; that paste mistake is rejected, since clients would otherwise hammer the replication listener. See the [multi-replica exposure callout](#service-exposure--session-affinity). |
| SASL credentials | struct\[] | — | list of `{username, password}` | `connectors.kafka.credentials` (config file / secret only) | — | **Secret — no env var, no CRD field.** SASL/PLAIN + SCRAM user store; when non-empty, SASL auth is enforced on every listener. `Validate()` rejects empty or duplicate usernames, empty passwords, and a username matching the reserved internal dashboard identity. Passwords are redacted in logs. |
**Kafka env-var trap: `CONNECTORS_KAFKA_MAX_TRANSACTIONAL_I_DS`.** The `MaxTransactionalIDs`
field renders to `..._MAX_TRANSACTIONAL_I_DS` (an extra underscore before `DS`), **not** the
intuitive `..._MAX_TRANSACTIONAL_IDS`. The wrong form does not bind and is silently ignored.
All other Kafka names follow the normal `CONNECTORS_KAFKA_*` rule.
### Kafka SASL credentials [#kafka-sasl-credentials]
**Kafka SASL credentials are secret/file-only.** Unlike AWS, there is **no `CredentialsData`
env-var escape hatch and no CRD field** — supply the `credentials` list through a mounted
`config.yaml` (or Secret-mounted file). On Kubernetes the delivery mechanism is
[`spec.envFromSecrets`](/configure/reference/deployment#advanced--other-top-level-spec-fields),
which projects an existing Secret into the pod without the values transiting the operator.
Plan the credential delivery path before enabling SASL on Kubernetes.
### OAUTHBEARER authentication [#oauthbearer-authentication]
OAUTHBEARER activates when `OAuthBearer.Issuer` is non-empty — there is no separate enable
flag. It is enforced only on the TLS/SASL\_SSL listener (`TlsPort`); like the advanced knobs
above, it is `config.yaml`/env-var-only (Helm/CRD path `—`, not a CRD field).
| Setting | Type | Default | Valid values | Docker (config.yaml key · env var) | Helm/CRD path | Notes |
| ----------------------------- | ------ | ------- | ------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------- | ------------- | ------------------------------------------------------------------------------------------------------------------------- |
| Issuer | string | `""` | non-empty to activate; else OAUTHBEARER stays off | `connectors.kafka.oauthbearer.issuer` · `CONNECTORS_KAFKA_OAUTH_BEARER_ISSUER` (alias of `CONNECTORS_KAFKAO_AUTH_BEARER_ISSUER`) | — | Non-empty **requires** a non-empty `TlsPort` — a bearer token must not cross a plaintext transport. |
| Client ID | string | `""` | required unless Skip client-ID check is `true` | `connectors.kafka.oauthbearer.clientid` (by convention) · `CONNECTORS_KAFKA_OAUTH_BEARER_CLIENT_ID` | — | The OIDC audience. Enforced only when `Issuer` is set. |
| Skip client-ID check | bool | `false` | true / false | `connectors.kafka.oauthbearer.skipclientidcheck` (by convention) · `CONNECTORS_KAFKA_OAUTH_BEARER_SKIP_CLIENT_ID_CHECK` | — | The one skip-flag Kafka **permits** as `true` — some IdPs legitimately omit or vary the audience claim. |
| Skip expiry check | bool | `false` | must stay `false` | `connectors.kafka.oauthbearer.skipexpirycheck` (by convention) · `CONNECTORS_KAFKA_OAUTH_BEARER_SKIP_EXPIRY_CHECK` | — | `true` is **hard-rejected** on the Kafka listener — stricter than the generic OIDC authentication path, which only warns. |
| Skip issuer check | bool | `false` | must stay `false` | `connectors.kafka.oauthbearer.skipissuercheck` (by convention) · `CONNECTORS_KAFKA_OAUTH_BEARER_SKIP_ISSUER_CHECK` | — | `true` is **hard-rejected** — it would accept tokens from any issuer. |
| Insecure skip signature check | bool | `false` | must stay `false` | `connectors.kafka.oauthbearer.insecureskipsignaturecheck` (by convention) · `CONNECTORS_KAFKA_OAUTH_BEARER_INSECURE_SKIP_SIGNATURE_CHECK` | — | `true` is **hard-rejected** — it would accept forged/unsigned tokens. |
**Both OAUTHBEARER env forms work — use the readable one.** The generic snake-caser fuses
`Kafka` and `OAuthBearer` into one word and produces the unguessable
`CONNECTORS_KAFKAO_AUTH_BEARER_*`. Because nobody can guess that, the server **also** binds
the natural **`CONNECTORS_KAFKA_OAUTH_BEARER_*`** form deliberately, and both resolve to the
same setting. Prefer the natural form:
```bash
CONNECTORS_KAFKA_OAUTH_BEARER_ISSUER=https://idp.example.com
CONNECTORS_KAFKA_OAUTH_BEARER_CLIENT_ID=kubemq
```
**One caveat: the server's unknown-variable warner does not know about the alias**, so
setting the natural form prints an `IGNORED` warning even though the value is applied. The
warning is wrong. Do not "fix" a working issuer because of it — confirm the effective value
in the dashboard instead.
**Skip-flags are hard-rejected on Kafka, not just warned.** Setting `Insecure skip signature
check`, `Skip issuer check`, or `Skip expiry check` to `true` on the Kafka listener is
**rejected outright** by `Validate()` — stricter than the generic OIDC authentication path,
which only warns. `Skip client-ID check` is the one permitted skip. (`OAuthBearer.Issuer` also
requires a non-empty `TlsPort` — see the Issuer row above.)
## AWS [#aws]
The AWS SQS/SNS-compatible connector. Env prefix `CONNECTORS_AWS_*`; CRD group `spec.aws.*`.
| Setting | Type | Default | Valid values | Docker (config.yaml key · env var) | Helm/CRD path | Notes |
| ------------------------ | --------- | -------------------- | -------------------------------------------------- | ------------------------------------------------------------------------------ | ------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Enable | bool | **`false` (opt-in)** | true / false | `connectors.aws.enable` · `CONNECTORS_AWS_ENABLE` | `spec.aws.enabled` | Opt-in wire connector. Opens port 4566. |
| Port | string | `4566` | `1`–`65535` | `connectors.aws.port` · `CONNECTORS_AWS_PORT` | `spec.aws.port` | Rejected if it collides with the gRPC/REST/HTTP or GCP listener port. |
| Region | string | `kubemq` | non-empty | `connectors.aws.region` · `CONNECTORS_AWS_REGION` | `spec.aws.region` | |
| Account ID | string | `000000000000` | `^[0-9]{12}$` | `connectors.aws.accountid` · `CONNECTORS_AWS_ACCOUNT_ID` | `spec.aws.accountId` | Exactly 12 digits. |
| Advertised URL | string | `""` | `""` or `scheme://host[:port]` | `connectors.aws.advertisedurl` · `CONNECTORS_AWS_ADVERTISED_URL` | `spec.aws.advertisedUrl` | Rejected unless it parses with a scheme and host. |
| Max inflight per queue | int | `20000` | `1`–`10000000` | `connectors.aws.maxinflightperqueue` · `CONNECTORS_AWS_MAX_INFLIGHT_PER_QUEUE` | `spec.aws.maxInflightPerQueue` | Over-ceiling rejected (OOM guard). |
| Max concurrent polls | int | `1024` | `1`–`1000000` | `connectors.aws.maxconcurrentpolls` · `CONNECTORS_AWS_MAX_CONCURRENT_POLLS` | `spec.aws.maxConcurrentPolls` | Over-ceiling rejected. |
| Read timeout (s) | int | `60` | `1`–`3600` | `connectors.aws.readtimeout` · `CONNECTORS_AWS_READ_TIMEOUT` | `spec.aws.readTimeout` | Per-action sync deadline; over-ceiling rejected. |
| Body limit | string | `2M` | size string | `connectors.aws.bodylimit` · `CONNECTORS_AWS_BODY_LIMIT` | `spec.aws.bodyLimit` | |
| Message signing | bool | `false` | true / false | `connectors.aws.messagesigning` · `CONNECTORS_AWS_MESSAGE_SIGNING` | `spec.aws.messageSigning` | Sign SNS Notification / SubscriptionConfirmation envelopes (SigV2). |
| Signing cert TTL (h) | int | `8760` | ≥ `1` | `connectors.aws.signingcertttlhours` · `CONNECTORS_AWS_SIGNING_CERT_TTL_HOURS` | `spec.aws.signingCertTtlHours` | Self-signed signing-cert validity (default 365 days). Applies when message signing is on. |
| Credentials (data blob) | string | `""` | JSON or base64-of-JSON credential array | `connectors.aws.credentialsdata` · `CONNECTORS_AWS_CREDENTIALS_DATA` | `spec.aws.credentialsData` | SigV4 credential array. On Helm/CRD this is rendered into a **Kubernetes Secret** (not a plain ConfigMap). |
| Credentials (structured) | struct\[] | — | list of `{accessKeyId, secretAccessKey, clientID}` | `connectors.aws.credentials` (config file only) | — | **No env var, no CRD field.** File/structured-only credential list; the env/CRD path is `credentialsData`. Empty or duplicate `accessKeyId` is rejected; `clientID` defaults to `accessKeyId`. |
**AWS credentials go through a Secret, not a plain CRD value.** The `spec.aws.credentialsData`
field exists but the operator writes it into a Kubernetes **Secret** (`CONNECTORS_AWS_CREDENTIALS_DATA`),
never a ConfigMap. On Docker, set `connectors.aws.credentialsdata` (JSON or base64-of-JSON) or
the `CONNECTORS_AWS_CREDENTIALS_DATA` env var. The fully-structured `connectors.aws.credentials`
list is config-file-only and has no env/CRD route.
## GCP Pub/Sub [#gcp-pubsub]
The Google Cloud Pub/Sub emulator connector (gRPC). Env prefix `CONNECTORS_GCP_*`; CRD group
`spec.gcp.*`.
| Setting | Type | Default | Valid values | Docker (config.yaml key · env var) | Helm/CRD path | Notes |
| ----------------------------- | ------ | -------------------- | ----------------------------- | -------------------------------------------------------------------------------------------- | ------------------------------------- | ------------------------------------------------------------------------------------------------------------------ |
| Enable | bool | **`false` (opt-in)** | true / false | `connectors.gcp.enable` · `CONNECTORS_GCP_ENABLE` | `spec.gcp.enabled` | Opt-in wire connector. Opens port 8085. |
| Port | string | `8085` | `1`–`65535` | `connectors.gcp.port` · `CONNECTORS_GCP_PORT` | `spec.gcp.port` | gRPC listener (Pub/Sub emulator convention). Rejected if it collides with the gRPC/REST/HTTP or AWS listener port. |
| Advertised endpoint | string | `""` | endpoint | `connectors.gcp.advertisedendpoint` · `CONNECTORS_GCP_ADVERTISED_ENDPOINT` | `spec.gcp.advertisedEndpoint` | Endpoint advertised to clients. |
| Max message bytes | int | `10485760` (10 MiB) | `1`–`1073741824` (1 GiB) | `connectors.gcp.maxmessagebytes` · `CONNECTORS_GCP_MAX_MESSAGE_BYTES` | `spec.gcp.maxMessageBytes` | Feeds the gRPC frame ceiling (value + 1 MiB); hard cap 1 GiB. |
| Default ack deadline (s) | int | `10` | `10`–`600` | `connectors.gcp.defaultackdeadlineseconds` · `CONNECTORS_GCP_DEFAULT_ACK_DEADLINE_SECONDS` | `spec.gcp.defaultAckDeadlineSeconds` | |
| Max outstanding messages | int | `1000` | > `0` | `connectors.gcp.maxoutstandingmessages` · `CONNECTORS_GCP_MAX_OUTSTANDING_MESSAGES` | `spec.gcp.maxOutstandingMessages` | |
| Max inflight per subscription | int | `20000` | > `0` | `connectors.gcp.maxinflightpersubscription` · `CONNECTORS_GCP_MAX_INFLIGHT_PER_SUBSCRIPTION` | `spec.gcp.maxInflightPerSubscription` | |
| Max concurrent polls | int | `1024` | > `0` | `connectors.gcp.maxconcurrentpolls` · `CONNECTORS_GCP_MAX_CONCURRENT_POLLS` | `spec.gcp.maxConcurrentPolls` | |
| Max concurrent streams | int | `1024` | `0` (default) or `1`–`65536` | `connectors.gcp.maxconcurrentstreams` · `CONNECTORS_GCP_MAX_CONCURRENT_STREAMS` | `spec.gcp.maxConcurrentStreams` | Per-server cap on concurrent StreamingPull streams. `0` = use the built-in default (1024). |
| Delivery shards | int | `16` | `1`–`256` | `connectors.gcp.deliveryshards` · `CONNECTORS_GCP_DELIVERY_SHARDS` | `spec.gcp.deliveryShards` | Striped delivery-pool shards. |
| Max ack extension (s) | int | `600` | `0` (disabled) or `10`–`3600` | `connectors.gcp.maxackextensionseconds` · `CONNECTORS_GCP_MAX_ACK_EXTENSION_SECONDS` | `spec.gcp.maxAckExtensionSeconds` | `0` disables the ordered-head ack-deadline keep-alive. |
| Stream close (s) | int | `1800` | > `0` | `connectors.gcp.streamcloseseconds` · `CONNECTORS_GCP_STREAM_CLOSE_SECONDS` | `spec.gcp.streamCloseSeconds` | |
| Max seek replay | int | `1000000` | > `0` | `connectors.gcp.maxseekreplay` · `CONNECTORS_GCP_MAX_SEEK_REPLAY` | `spec.gcp.maxSeekReplay` | |
| Enable reflection | bool | `false` | true / false | `connectors.gcp.enablereflection` · `CONNECTORS_GCP_ENABLE_REFLECTION` | `spec.gcp.enableReflection` | gRPC server reflection. |
**Enable GCP Pub/Sub explicitly.** A stock kubemq-server does **not** bind port 8085 until you set
`CONNECTORS_GCP_ENABLE=true` (Docker) or `spec.gcp.enabled: true` (Kubernetes). Point clients at
the connector with `PUBSUB_EMULATOR_HOST=localhost:8085` — no auth, no TLS (emulator mode).