# Connect Kafka (/docs/guides/ingest/kafka-connector)



Connect Kafka [#connect-kafka]

The connector runs as a RawTree Cloud or RawTree BYOC service component, where enabled.
RawTree stores and queries the rows that it delivers.

Use a Kafka connector to read selected topics and write their messages to a
RawTree table. The connector service must be enabled for your environment.
Send API requests to `https://api.rawtree.com`.

Before you begin [#before-you-begin]

Create a destination database and table. Save the organization name, cluster
name, database name, and table name. Ask an organization administrator to
create, update, or delete the connector.

Prepare broker TLS certificates and SASL credentials when your broker requires
them. Do not place private keys or passwords in shared shell history.

Preview a broker [#preview-a-broker]

Send a preview request before saving the connector. For an unsaved connector,
use `POST /v1/connectors/preview`. The preview can list topics or sample up to
100 messages from one exact topic.

```json
{
  "type": "kafka",
  "settings": {
    "bootstrap_servers": "broker.example.com:9093",
    "tls": {
      "enabled": true,
      "verify_certificate": true,
      "verify_hostname": true,
      "ca_certificate": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----"
    },
    "sasl": {
      "enabled": true,
      "mechanism": "SCRAM-SHA-256",
      "username": "rawtree-reader",
      "password": "replace-with-a-secret"
    }
  },
  "operation": "list_topics"
}
```

The preview response includes the observed time and topic results. Save the
topic name you want to ingest. A saved connector uses
`POST /v1/connectors/{connector_id}/preview` to preview with stored credentials.

Create a connector [#create-a-connector]

Send `POST /v1/connectors?organization=<organization>&cluster=<cluster>`.
`settings` accepts TLS, SASL, and batch options. The service stores credentials
encrypted and omits them from the response.

```json
{
  "name": "product-events",
  "type": "kafka",
  "destinations": [
    {
      "topics": ["product.events"],
      "group_id": "rawtree-product-events",
      "database": "analytics",
      "table": "product_events"
    }
  ],
  "settings": {
    "bootstrap_servers": "broker.example.com:9093",
    "auto_offset_reset": "smallest",
    "tls": {
      "enabled": true,
      "verify_certificate": true,
      "verify_hostname": true,
      "ca_certificate": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----"
    },
    "sasl": {
      "enabled": true,
      "mechanism": "SCRAM-SHA-256",
      "username": "rawtree-reader",
      "password": "replace-with-a-secret"
    },
    "batch": {
      "max_events": 1000,
      "timeout_secs": 1
    }
  }
}
```

The API returns the connector ID, status, revision, and runtime state. Save
`id` as `connector_id`. Check `runtime.state` and `runtime.checked_at` to see
whether the service has checked the connector.

Check topics or messages [#check-topics-or-messages]

Call `POST /v1/connectors/{connector_id}/preview` with one operation:

```json
{ "operation": "list_topics" }
```

To sample messages, send an exact topic name and a limit from 1 to 100:

```json
{
  "operation": "preview_topic",
  "parameters": {
    "topic": "product.events",
    "limit": 10
  }
}
```

Listing topics needs broker Describe permission. Sampling also needs Read
permission. A preview has a ten-second deadline and does not change offsets.

Check connector metrics [#check-connector-metrics]

Use a first-party session or OAuth access token to read connector telemetry
through `GET /v1/metrics`. This endpoint does not accept API keys. Filter by the
saved connector ID:

```text
GET /v1/metrics?organization=acme&cluster=analytics&metric=rawtree.connector.events.delivered&filters=%7B%22rawtree.connector.id%22%3A%22<connector_id>%22%7D
```

Save the returned metric series and timestamps. The API has no dedicated
`/v1/connectors/{connector_id}/metrics` route. See
[Monitor usage](/docs/guides/operate/monitor-usage) for supported metrics.

Update or delete a connector [#update-or-delete-a-connector]

Send only the fields to change in `PATCH /v1/connectors/{connector_id}`. Save
the returned `revision` and confirm the new `runtime.state`.

```json
{
  "status": "paused",
  "settings": {
    "bootstrap_servers": "broker.example.com:9093",
    "auto_offset_reset": "largest",
    "tls": { "enabled": true },
    "sasl": {
      "enabled": true,
      "mechanism": "SCRAM-SHA-256",
      "username": "rawtree-reader",
      "password": "replace-with-a-secret"
    }
  }
}
```

To remove the connector, send `DELETE /v1/connectors/{connector_id}`. The
response contains `deleted: true`.