# CE ↔ KubeMQ Mapping (/connectors/cloudevents/reference/ce-to-kubemq-mapping)



The CloudEvents connector is a faithful, round-tripping bridge: every CloudEvent
attribute becomes a prefixed KubeMQ tag on the way in, and is reconstructed back
into a CloudEvent on the way out. This page is the authoritative reference for that
mapping — the attribute table, channel and `ClientID` resolution, how outbound
messages are detected as CloudEvents, the `data` vs `data_base64` rule, and the
error surface.

## Attribute mapping [#attribute-mapping]

When a CloudEvent is received, each of its attributes is stored as a KubeMQ message
tag with a `ce_` prefix. The `ce_` prefix is what makes the mapping reversible: any
subscriber — including [CESQL routing](/connectors/cloudevents/how-to/cesql-routing) —
can read the original CloudEvent attributes off the message tags.

| CloudEvent attribute | KubeMQ tag key       | Required             | Notes                                                                                                      |
| -------------------- | -------------------- | -------------------- | ---------------------------------------------------------------------------------------------------------- |
| `specversion`        | `ce_specversion`     | Yes                  | Always `"1.0"`. Its presence is the marker the connector uses to detect a CloudEvent on outbound delivery. |
| `type`               | `ce_type`            | Yes                  | Application-defined event type.                                                                            |
| `source`             | `ce_source`          | Yes                  | Also used as the KubeMQ `ClientID` (see [ClientID resolution](#clientid-resolution)).                      |
| `id`                 | `ce_id`              | Yes (auto-generated) | Also used as the `EventID` / `RequestID` / `MessageID`.                                                    |
| `subject`            | `ce_subject`         | No                   | Primary [channel resolution](#channel-resolution) source.                                                  |
| `time`               | `ce_time`            | No (auto-generated)  | RFC3339Nano format.                                                                                        |
| `datacontenttype`    | `ce_datacontenttype` | No                   | e.g. `application/json`.                                                                                   |
| `dataschema`         | `ce_dataschema`      | No                   | URI of the data schema.                                                                                    |
| *(any extension)*    | `ce_{name}`          | No                   | Extension attributes get the same `ce_` prefix.                                                            |

<Callout type="info">
  Extension attributes follow the same convention: `{"myextension": "value"}` is stored
  as the `ce_myextension` tag and reconstructed as `{"myextension": "value"}` on delivery.
</Callout>

## Auto-generation [#auto-generation]

The connector fills in missing optional attributes **before** validation, so a minimal
CloudEvent with only `type` and `source` is accepted:

| Attribute | When missing | Generated value                           |
| --------- | ------------ | ----------------------------------------- |
| `id`      | Empty        | A new UUID v4.                            |
| `time`    | Zero         | The current UTC time, RFC3339Nano format. |

Because `id` is always populated, every accepted CloudEvent has a stable identifier
for correlation and replay.

## Channel resolution [#channel-resolution]

The KubeMQ destination channel is resolved in priority order:

1. **CE `subject` attribute** — if the CloudEvent carries a `subject`, it is the channel name.
2. **`?channel=` query parameter** — used only when no `subject` is set.

If neither is provided, the request is rejected with **HTTP 400** (`channel is required`).

```bash
# subject sets the channel
curl -X POST http://localhost:9090/ce/send/event \
  -H "Content-Type: application/cloudevents+json" \
  -d '{"specversion":"1.0","type":"com.example.order.created","source":"order-service","subject":"orders","data":{"id":"123"}}'

# or the query parameter sets it (no subject)
curl -X POST "http://localhost:9090/ce/send/event?channel=orders" \
  -H "Content-Type: application/cloudevents+json" \
  -d '{"specversion":"1.0","type":"com.example.order.created","source":"order-service","data":{"id":"123"}}'
```

See [Channel resolution](/connectors/cloudevents/how-to/channel-resolution) for the
full priority rules and SSE subscription behavior.

## ClientID resolution [#clientid-resolution]

The KubeMQ `ClientID` attached to the message is determined as follows:

1. **Auth claims** — when [authentication](/connectors/cloudevents/how-to/authentication)
   is enabled and the request carries valid credentials, the authenticated `ClientID`
   from the JWT claims overrides everything else.
2. **CE `source` attribute** — used as the `ClientID` when auth is off or the claim is `anonymous`.

This means the CloudEvent `source` is the effective identity for unauthenticated
traffic, while authenticated traffic always carries its verified identity regardless
of what `source` says.

## Outbound CE detection [#outbound-ce-detection]

When delivering a message to a subscriber — over [SSE](/connectors/cloudevents/how-to/sse-behavior)
or a [queue receive](/connectors/cloudevents/how-to/queues) — the connector
inspects the message tags for `ce_specversion`:

| Condition                    | Delivered as                                                                           | SSE `event:` type |
| ---------------------------- | -------------------------------------------------------------------------------------- | ----------------- |
| `ce_specversion` **present** | A reconstructed CloudEvent JSON object with standard CE attributes.                    | `cloudevent`      |
| `ce_specversion` **absent**  | A plain JSON object with KubeMQ-native fields (`channel`, `metadata`, `tags`, `data`). | `message`         |

Because detection is per-message, a single channel can carry **mixed** CloudEvent and
non-CloudEvent traffic — each frame is shaped according to its own tags. Clients
subscribing to such a channel should handle both `cloudevent` and `message` event types.

## data vs data\_base64 [#data-vs-data_base64]

When reconstructing a CloudEvent for outbound delivery, the connector inspects the
message body:

| Body                                       | Carried in    | Encoding               |
| ------------------------------------------ | ------------- | ---------------------- |
| Valid JSON                                 | `data`        | Inline JSON object.    |
| Binary or non-JSON (plain text, raw bytes) | `data_base64` | Base64-encoded string. |

This follows the CloudEvents JSON format specification, where `data_base64` is the
standard carrier for non-JSON payloads.

## CESQL attribute names [#cesql-attribute-names]

[CESQL routing](/connectors/cloudevents/how-to/cesql-routing) expressions reference
CloudEvent attribute names **without** the `ce_` prefix, even though the underlying
KubeMQ tags carry it:

```sql
type = 'com.example.order.created'   -- matches the ce_type tag
source = 'order-service'             -- matches the ce_source tag
```

The router constructs a lightweight CloudEvent from the `ce_*` tags at evaluation time,
so any message with `ce_*` tags — regardless of which connector produced it — is eligible
for CESQL matching.

## Error codes [#error-codes]

All CloudEvents endpoints return a single response envelope. Errors use the same shape
with `is_error: true`:

```json
{
  "is_error": true,
  "message": "descriptive error message",
  "data": null
}
```

### HTTP status codes [#http-status-codes]

| Status | Meaning               | When                                                                                                                                                  |
| ------ | --------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- |
| 200    | OK                    | Query response, queue receive, queue ack\_all.                                                                                                        |
| 202    | Accepted              | Event send, event-store send, command, queue send, response send.                                                                                     |
| 400    | Bad Request           | Invalid CloudEvent (including unrecognized Content-Type), missing required parameters, validation failure, reserved channel name, subscription error. |
| 429    | Too Many Requests     | SSE connection limit (`MaxSSEConnections`) exceeded.                                                                                                  |
| 500    | Internal Server Error | Backend messaging error or SSE setup failure.                                                                                                         |
| 504    | Gateway Timeout       | A synchronous request exceeded `TimeoutSeconds`.                                                                                                      |

### Common error messages [#common-error-messages]

| Message                     | Cause                                                                                 | Resolution                                                  |
| --------------------------- | ------------------------------------------------------------------------------------- | ----------------------------------------------------------- |
| `channel is required`       | No `subject` attribute and no `?channel=` parameter.                                  | Set the CloudEvent `subject` attribute or pass `?channel=`. |
| `invalid CloudEvent`        | Malformed CE JSON or a missing required attribute.                                    | Ensure `specversion`, `type`, and `source` are present.     |
| `stream idle timeout`       | No messages delivered for `MaxSSEIdleSeconds`.                                        | Reconnect; consider lowering `MaxSSEIdleSeconds`.           |
| `context deadline exceeded` | Request exceeded `TimeoutSeconds`.                                                    | Increase the timeout or ensure a responder is active.       |
| `connection limit exceeded` | `MaxSSEConnections` reached.                                                          | Raise the limit or reduce concurrent SSE connections.       |
| `reserved channel name`     | Channel name conflicts with an internal KubeMQ channel (e.g. the `_AGENTS_.` prefix). | Use a different channel name.                               |

## Related [#related]

<Cards>
  <Card title="Endpoints" href="/connectors/cloudevents/reference/endpoints" description="Full endpoint table, query parameters, and status codes." />

  <Card title="Content modes" href="/connectors/cloudevents/how-to/content-modes" description="Structured vs binary mode and how each maps to tags." />

  <Card title="Channel resolution" href="/connectors/cloudevents/how-to/channel-resolution" description="Subject, query-parameter, and ClientID resolution in depth." />

  <Card title="CESQL routing" href="/connectors/cloudevents/how-to/cesql-routing" description="Route messages by CloudEvent attribute with CESQL expressions." />
</Cards>