# Audit Logs

Consumes **Confluent Cloud audit log events** (CloudEvents 1.0) from the dedicated `confluent-audit-log-events` Kafka topic on Confluent's audit log cluster.

**Sync Type: Streaming**

## Details

Confluent Cloud delivers organization-wide audit log events to a **dedicated Kafka cluster** (separate from any of your data clusters), on the fixed topic `confluent-audit-log-events`. Confluent does not expose a REST API for audit logs — Kafka consumption is the only first-party programmatic access mechanism.

This connector consumes that topic and forwards each event into the pipeline. Each record is a [CloudEvents 1.0](https://github.com/cloudevents/spec) JSON object with a Confluent-specific `data` payload covering categories like authentication, authorization, ACL management, and topic management.

The audit topic has a **7-day retention** on Confluent's side. The default `Start Offset` is `earliest` so a fresh pipeline picks up everything currently available; once a consumer group has committed offsets, subsequent runs resume from the committed position.

## Prerequisites

Before configuring this input you need:

1. **Audit Log enabled on your Confluent Cloud organization.** Audit Log is automatically provisioned for paid Confluent Cloud orgs; verify via the Console under **Administration → Audit Log**, or run `confluent audit-log describe`.

2. **Audit cluster bootstrap server.** The audit log cluster is distinct from your data clusters. Find its bootstrap endpoint in **Administration → Audit Log → Cluster details**, or via `confluent audit-log describe` (look for the `bootstrap_servers` field). It looks like `lkc-xxxxxx.region.provider.confluent.cloud:9092`.

3. **An API Key + Secret issued against the audit log cluster.** This is critical — keys against your normal data clusters will not authenticate. The key must be created by a user holding the **OrganizationAdmin** role (or by an automation that role has authorized).

4. **A consumer group ID** for Monad (e.g. `monad-confluent-audit-logs`).

5. **Network reachability** to the audit cluster on port `9092` (SASL_SSL).

## Setup Instructions

### Step 1: Confirm Audit Log is provisioned

In the Confluent Cloud Console, sign in as an OrganizationAdmin and open **Administration → Audit Log**. The page lists the audit cluster ID, bootstrap server, and topic name (`confluent-audit-log-events`).

Alternatively, from the Confluent CLI:

```bash
confluent audit-log describe
```

### Step 2: Create an API Key against the audit cluster

Audit-cluster API keys can only be issued by an OrganizationAdmin.

Via the Console:

1. Go to **Administration → Audit Log → Audit log cluster → API keys**.
2. Click **Create key**, pick a Service account, and create the key.
3. Copy the **Key** and **Secret** — the Secret is shown only once.

Via the CLI:

```bash
confluent api-key create --resource <audit-cluster-id> --service-account <sa-id>
```

### Step 3: Configure the connector in Monad

1. **Bootstrap Servers** — paste the audit-cluster bootstrap endpoint from Step 1.
2. **Topic** — leave as `confluent-audit-log-events` (the default). Only change this if Confluent has reconfigured your audit cluster.
3. **Consumer Group** — a dedicated ID for Monad, e.g. `monad-confluent-audit-logs`.
4. **API Key** / **API Secret** — the values from Step 2.
5. **Start Offset** — `earliest` is recommended. The audit topic only retains 7 days of events on Confluent's side, so `earliest` lets you backfill that window on first run.
6. **Message Format** — `json` (audit records are CloudEvents JSON).

The connector always authenticates as **SASL_SSL / PLAIN** — Confluent's only supported mode for the audit cluster — and uses your platform's system trust store for TLS verification.

## Configuration

### Settings

| Setting | Type | Default | Description |
|---------|------|---------|-------------|
| Bootstrap Servers | array of strings | — | Bootstrap server(s) for the Confluent Cloud audit log cluster. Find under Console → Administration → Audit Log, or `confluent audit-log describe`. |
| Topic | string | `confluent-audit-log-events` | The audit log topic. Confluent fixes this name; only change it if Confluent has reconfigured your audit cluster. |
| Consumer Group | string | — | Consumer group ID Monad uses to track offsets. Use a dedicated ID. |
| API Key | string | — | Confluent Cloud API Key for the **audit log cluster** (used as the SASL username). Must be created by an OrganizationAdmin. |
| API Secret | secret | — | Confluent Cloud API Secret paired with the audit log cluster API Key. |
| Start Offset | string | `earliest` | Where to start when no committed offset exists for the consumer group. Options: `earliest`, `latest`. |
| Message Format | string | `json` | Audit records are CloudEvents JSON; leave as `json`. |

The connector authenticates as **SASL_SSL / PLAIN** unconditionally — no security-protocol selector, no SASL-mechanism selector, no CA Certificate upload.

## Troubleshooting

### Authentication failure / `SaslAuthenticationException`

- Confirm the API Key was created **against the audit log cluster**, not a regular data cluster. Keys are cluster-scoped.
- Confirm the key was issued by an OrganizationAdmin — only that role can create audit-cluster keys.
- The **SASL Mechanism** must be `PLAIN`.

### Connection refused / DNS errors

- Re-fetch the bootstrap endpoint from **Administration → Audit Log** or `confluent audit-log describe`. The audit cluster bootstrap hostname differs from your data clusters and changes if Confluent migrates your audit cluster.
- Make sure outbound port `9092` is open from Monad's network to `*.confluent.cloud`.

### `Group Authorization Failed` (error code 30)

```
[30] Group Authorization Failed: the client is not authorized to access a particular group id
```

In Confluent Cloud, the consumer group ID is its own ACL resource — separate from the topic. Test Connection succeeds (broker auth + topic metadata only), but consumption requires an explicit ACL on the group.

Grant the audit-cluster API key's service account `READ` on the Consumer Group ID you set in Monad:

```bash
confluent kafka acl create \
  --allow \
  --service-account <sa-id> \
  --operations read \
  --consumer-group <your-consumer-group-id> \
  --cluster <audit-cluster-id>
```

Or via the Console: **Administration → Audit Log → Audit log cluster → ACLs → Add ACL**.

### Topic not found

- Leave **Topic** as `confluent-audit-log-events`. Audit events do not flow to any other topic.
- If the topic truly does not exist on the audit cluster, audit logging may not be provisioned for your organization — file a support request with Confluent.

### Missing or stale events

- The audit cluster retains only **7 days** of events. Older events cannot be recovered from Confluent. If you need long-term retention, run this connector continuously and warehouse the output downstream.
- If **Start Offset** is `latest` and the consumer group is brand new, Monad will only forward events written after the pipeline starts. Use `earliest` to backfill the available window.
- Ensure no other consumer is sharing the same Consumer Group ID — if it is, that consumer commits offsets and the audit events will appear split across the two consumers.

## Best Practices

- **Dedicated service account.** Create a service account specifically for Monad's audit log consumption and issue the API key against it. Rotate the secret on the cadence required by your security policy (Confluent recommends monthly for high-security environments).
- **Run continuously.** Because Confluent only keeps 7 days of audit events, a long pipeline outage risks data loss. Monitor pipeline health and alert on extended downtime.
- **Treat audit logs as a long-retention destination.** Ship the connector's output to a SIEM or warehouse with retention measured in years, not days.
- **Don't share the Consumer Group** across consumers. The audit topic is read-only; double-consumption (two pipelines using the same group) splits partitions and each consumer misses events the other handled.

## Sample Record

A representative audit event (authentication category):

```json
{
  "id": "29ca0e51-fdcd-44bd-a393-43193432b614",
  "source": "crn://confluent.cloud/kafka=lkc-6k8r8q",
  "specversion": "1.0",
  "type": "io.confluent.kafka.server/authentication",
  "datacontenttype": "application/json",
  "subject": "crn://confluent.cloud/kafka=lkc-6k8r8q",
  "time": "2026-06-17T22:41:43.395Z",
  "data": {
    "serviceName": "crn://confluent.cloud/kafka=lkc-6k8r8q",
    "methodName": "kafka.Authentication",
    "resourceName": "crn://confluent.cloud/kafka=lkc-6k8r8q",
    "authenticationInfo": {
      "principal": "User:306343",
      "metadata": {
        "mechanism": "SASL_SSL/PLAIN",
        "identifier": "MAIDSRFG53RXYTKR"
      },
      "principalResourceId": "u-yw9507",
      "identity": "crn://confluent.cloud/organization=uuid/identity-provider=idp/identity=u-yw9507"
    },
    "result": {
      "status": "SUCCESS",
      "message": ""
    }
  }
}
```

## Related Articles

- [Confluent Cloud — Audit Log overview](https://docs.confluent.io/cloud/current/monitoring/audit-logging/)
- [Confluent Cloud — Audit Log concepts](https://docs.confluent.io/cloud/current/monitoring/audit-logging/cloud-audit-log-concepts.html)
- [Confluent Cloud — Configure access to audit logs](https://docs.confluent.io/cloud/current/monitoring/audit-logging/configure.html)
- [Confluent Cloud — Audit log event categories](https://docs.confluent.io/cloud/current/monitoring/audit-logging/cloud-audit-log-categories.html)
- [Confluent Cloud — Audit log best practices](https://docs.confluent.io/cloud/current/monitoring/audit-logging/best-practices.html)
- [Confluent Cloud — Service accounts](https://docs.confluent.io/cloud/current/security/authenticate/workload-identities/service-accounts/manage-service-accounts.html)
- [CloudEvents 1.0 specification](https://github.com/cloudevents/spec/blob/v1.0.2/cloudevents/spec.md)
- [Monad — Confluent Messages input](./confluent-messages)