> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cognite.com/llms.txt
> Use this file to discover all available pages before exploring further.

# MQTT broker

> Learn how to publish data directly to a Cognite-managed MQTT broker and ingest it into Cognite Data Fusion (CDF).

<Warning>
  The features described in this section are in [public preview](/cdf/product_feature_status#public-preview) and may change.
</Warning>

The Cognite MQTT broker is an **inbound** endpoint — your systems publish data directly to a Cognite-managed MQTT broker. This is the reverse of the [Cognite MQTT extractor](/cdf/integration/guides/extraction/mqtt), where Cognite connects as a subscriber to your broker.

Use the MQTT broker when your systems can push data over MQTT but you don't want to run and expose your own broker.

## Before you start

* Assign [access capabilities](/cdf/access/guides/capabilities#cognite-mqtt-extractor) to *create* an MQTT broker source and for the extractor to *write* data points, time series, events, RAW rows, and instances in data models to the target CDF project, depending on the resource types you want the extractor to ingest.
* Before you create the job, reuse or create a [destination](/api-reference/concepts/20230101/hosted-extractors-destinations). If you use a custom message format, reuse or create a [mapping](/api-reference/concepts/20230101/hosted-extractors-mappings) as described in [Custom message formats](/cdf/integration/guides/extraction/hosted_extractors/hosted_extractors_custom_mappings).
* The MQTT broker must be enabled for your project. Contact Cognite support to request access.
* The Cognite MQTT broker is not available for Private Link–enabled CDF projects.

<Note>
  This setup uses two different credentials:

  * **CDF API** (create the source and job): use your usual CDF authentication, such as OpenID Connect (OIDC), as described in [setup and administration for extractors](/cdf/integration/guides/extraction/admin_oidc).
  * **MQTT publisher** (send data): use the `host`, `username`, and `password` returned when you create the source.
</Note>

## Set up the MQTT broker

<Steps>
  <Step title="Create a source">
    Create a `Source` by sending a request to `/api/v1/projects/{project}/hostedextractors/sources`. Include the `cdf-version: beta` header. For the full request schema, see the [Source API specification](/api-reference/concepts/20230101/hosted-extractors-sources).

    ```json theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
    {
      "items": [{
        "type": "mqtt_broker",
        "externalId": "my-mqtt-broker-source"
      }]
    }
    ```

    The response contains the connection details your publisher needs:

    ```json theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
    {
      "items": [{
        "type": "mqtt_broker",
        "externalId": "my-mqtt-broker-source",
        "host": "<cognite-managed-host>",
        "port": 8883,
        "authentication": {
          "type": "basic",
          "username": "<generated-username>",
          "password": "<generated-password>"
        }
      }]
    }
    ```

    <Warning>
      The password is returned only when you create the source. Save the password by storing it immediately. If you read the source again later, the API will not return the password. If you lose the password, reset it as described in [Managing credentials](#managing-credentials).
    </Warning>
  </Step>

  <Step title="Create a job">
    Create a `Job` by sending a request to `/api/v1/projects/{project}/hostedextractors/jobs`. The job must reference your destination and mapping. For the full request schema, see the [Job API specification](/api-reference/concepts/20230101/hosted-extractors-jobs).

    Set `topicFilter` to match the topics your publisher will send data on.

    ```json theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
    {
      "items": [{
        "externalId": "my-mqtt-broker-job",
        "sourceId": "my-mqtt-broker-source",
        "format": {
          "type": "custom",
          "mappingId": "my-custom-mapping"
        },
        "destinationId": "my-destination",
        "config": {
          "topicFilter": "company/+/sensors/#"
        }
      }]
    }
    ```

    Messages must be transformed into a CDF resource type. For supported formats and custom mappings, see [Message formats](#message-formats).

    <Check>
      A `200` response with `connected` or `running` in the response body confirms that you created the job.
    </Check>

    If the [job](/cdf/integration/guides/extraction/hosted_extractors#jobs) is not `connected` or `running`, work through the checks below.

    <AccordionGroup>
      <Accordion title="Job not connected or running">
        You may see one of these responses:

        * `200` with `connection_error` or `startup_error` in the body
        * `400 Bad Request` or `422 Unprocessable Entity`

        Use the [Job API specification](/api-reference/concepts/20230101/hosted-extractors-jobs) to identify, decode, and resolve specific error codes. Also verify the following.

        * Your authentication token is valid.
        * Your request payload matches the request schema.
      </Accordion>
    </AccordionGroup>
  </Step>

  <Step title="Connect your publisher">
    Configure your MQTT client to connect using the credentials from step 1:

    * **Host**: the `host` value from the response in step 1
    * **Port**: `8883`
    * **Username**: the `authentication.username` value
    * **Password**: the `authentication.password` value
    * **TLS**: required. See [Encryption](#encryption) for certificate requirements.

    <Check>
      A success message or response state in your MQTT client confirms that your publisher connected to the broker.
    </Check>

    If your publisher does not connect, work through the checks below.

    <AccordionGroup>
      <Accordion title="Publisher not connected">
        Your MQTT client may show an error message or a failed connection state. Use your MQTT client documentation to identify, decode, and resolve specific error codes, then try connecting again. Also verify the following.

        * The host, port, username, and password match the values from step 1.
        * TLS is enabled and your device trusts the broker certificate. See [Encryption](#encryption).
        * You saved the password from step 1 and have not reset it since.
      </Accordion>
    </AccordionGroup>
  </Step>
</Steps>

## Managing credentials

The password can be reset at any time by sending a `POST` request to `/api/v1/projects/{project}/hostedextractors/sources/update` with the `cdf-version: beta` header:

```json theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
{
  "items": [{
    "type": "mqtt_broker",
    "externalId": "my-mqtt-broker-source",
    "update": {
      "password": { "reset": true }
    }
  }]
}
```

The response contains the new password. The previous password is immediately invalidated — connected publishers will be disconnected and must reconnect with the new credentials.

## Topics and topic filters

Messages are published to **topics** — hierarchical strings separated by forward slashes:

```text theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
company/location3/machine1/pressure
```

Jobs listen for messages using *topic filters*. A topic filter can be an exact topic or a wildcard pattern, allowing a single job to receive data from multiple topics. MQTT supports single-level (`+`) and multi-level (`#`) wildcards.

* `+` matches a single level. `company/+/pressure` matches `company/machine1/pressure` and `company/machine2/pressure`.
* `#` matches all remaining levels and must appear at the end. `company/location3/#` matches all topics under `company/location3/`.

Custom mappings receive the arrived topic in `context.topic`. See [Message formats](#message-formats) for examples of `input` and `context` arguments.

## Message formats

The Cognite MQTT broker supports [pre-defined message formats](/cdf/integration/guides/extraction/hosted_extractors/hosted_extractors_mappings) and [custom message formats](/cdf/integration/guides/extraction/hosted_extractors/hosted_extractors_custom_mappings).

Custom mappings receive two arguments: `input`, which contains the message payload as JSON, and `context`, which contains the topic the message arrived on.

For example, when a publisher sends `{"pressure": 42.1}` on topic `company/location3/machine1/pressure`, the mapping receives the following values.

**input** (message payload):

```json theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
{
  "pressure": 42.1
}
```

**context** (arrival metadata):

```json theme={"languages":{"custom":["/_languages/kuiper.json","../_languages/kuiper.json"]}}
{
  "topic": "company/location3/machine1/pressure"
}
```

The topic in `context` is the actual topic the message arrived on, not the topic filter. Use `context.topic` to distinguish messages arriving on different topics.

## Encryption

All connections to the Cognite MQTT broker require Transport Layer Security (TLS). The broker certificate is signed by [Let's Encrypt](https://letsencrypt.org/certificates/). Let's Encrypt root certificates are included in most modern operating systems — make sure the device running your MQTT client has up-to-date certificates.