> For the complete documentation index, see [llms.txt](https://docs.e6data.com/ingestion-engine/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.e6data.com/ingestion-engine/connectors/message-queues/confluent.md).

# Confluent

**Source** | **Sink**

The Confluent connector is a convenience wrapper around the [Kafka connector](/ingestion-engine/connectors/message-queues/kafka.md). It connects to Kafka clusters hosted on Confluent Cloud, but replaces the raw SASL authentication config with Confluent's simpler API key / secret pair. Under the hood, e6 Ingestion Engine converts a Confluent connection into a Kafka connection with SASL/PLAIN over SSL - so the same Kafka source and sink behavior applies, and you get the same exactly-once guarantees, checkpoint-based recovery, and rdkafka tuning options.

If you're running self-hosted Kafka, Amazon MSK, or any other non-Confluent broker, use the [Kafka connector](/ingestion-engine/connectors/message-queues/kafka.md) directly.

***

## Prerequisites

Before configuring the connector, make sure the following are in place on the Confluent Cloud side.

### Confluent Cloud Resources

* An active Confluent Cloud environment with a Kafka cluster in your target region
* Topics created and receiving data - e6 Ingestion Engine consumes from existing topics, it does not create them
* Schema Registry enabled on the environment (only if using Avro, Protobuf, or SR-JSON)

### Credentials

e6 Ingestion Engine needs up to two independent API key pairs. Both are static, customer-issued secrets with no automatic expiry.

* **Cluster API key + secret** (required) - Scoped to the specific Kafka cluster. Find it under **Cluster > API Keys > Create Key**. e6 Ingestion Engine maps the key to SASL `username` and the secret to SASL `password` over SASL\_SSL.
* **Schema Registry API key + secret** (optional) - Scoped to the Confluent Cloud environment, not the cluster. A single SR key can serve multiple clusters in the same environment. Find it under **Environment > Schema Registry > API credentials**.

### Service Account ACLs

Create a dedicated service account in Confluent Cloud for e6 Ingestion Engine. Do not reuse a human user's API key. Grant the least-privilege set:

**Source (consuming from a topic):**

* `READ` + `DESCRIBE` on the topic
* `READ` on the consumer group (see [Consumer Group](/ingestion-engine/connectors/message-queues/kafka.md#consumer-group) for naming)
* `DESCRIBE` on the cluster

**Sink (producing to a topic) - add to the above:**

* `WRITE` + `DESCRIBE` on the sink topic
* `IDEMPOTENT_WRITE` on the cluster (for `at_least_once` and `exactly_once` modes)
* `WRITE` + `DESCRIBE` on the transactional ID (only for `exactly_once` mode)

If your org uses Confluent RBAC instead of classic ACLs, the equivalent role bindings are `DeveloperRead` on the topic + consumer group (source) and `DeveloperWrite` on the topic (sink).

### Network Access

e6 Ingestion Engine initiates outbound connections to Confluent Cloud. No inbound access from Confluent is required.

{% tabs %}
{% tab title="Azure" %}

* **Kafka bootstrap**: outbound TCP port `9092` (SASL\_SSL) from the e6 Ingestion Engine worker pods to the Confluent Cloud cluster endpoint
* **Schema Registry**: outbound HTTPS port `443` to the Schema Registry endpoint
* If the Confluent Cloud cluster uses IP filtering, allowlist the NAT Gateway public IPs of the e6data compute plane
* Confluent Cloud Dedicated clusters with PrivateLink-only (no public endpoint) are not currently supported - the cluster must be reachable over the public internet. As a workaround, enable a public endpoint alongside the private one and use IP filtering.
  {% endtab %}

{% tab title="AWS" %}

* **Kafka bootstrap**: outbound TCP port `9092` (SASL\_SSL) from the e6 Ingestion Engine worker pods to the Confluent Cloud cluster endpoint
* **Schema Registry**: outbound HTTPS port `443` to the Schema Registry endpoint
* If the Confluent Cloud cluster uses IP filtering, allowlist the NAT Gateway public IPs of the e6data compute plane
* For Confluent Cloud Dedicated clusters with AWS PrivateLink, work with your e6data SE to configure VPC endpoint connectivity
  {% endtab %}

{% tab title="GCP" %}

* **Kafka bootstrap**: outbound TCP port `9092` (SASL\_SSL) from the e6 Ingestion Engine worker pods to the Confluent Cloud cluster endpoint
* **Schema Registry**: outbound HTTPS port `443` to the Schema Registry endpoint
* If the Confluent Cloud cluster uses IP filtering, allowlist the NAT Gateway public IPs of the e6data compute plane
  {% endtab %}
  {% endtabs %}

### Secret Rotation

Confluent Cloud API keys have no automatic expiry. Rotation is your responsibility. When rotating:

1. Create a new key in Confluent Cloud
2. Update the key in the e6 Ingestion Engine connection config
3. Confirm the pipeline reconnects successfully with the new key
4. Only then delete the old key in Confluent Cloud - active connections continue using the old key until they reconnect, so deleting it prematurely will break in-flight consumers

***

## Connection

The connection config holds the Confluent Cloud cluster address and API credentials. It is shared across all sources and sinks that target the same cluster.

```yaml
bootstrap_servers: pkc-xxxxx.us-east-1.aws.confluent.cloud:9092
key: "${CONFLUENT_API_KEY}"
secret: "${CONFLUENT_API_SECRET}"
```

* **`bootstrap_servers`** - Confluent Cloud bootstrap server address. Find this in your Confluent Cloud cluster settings under **Cluster Overview > Cluster Settings > Bootstrap server**. e6 Ingestion Engine uses it for initial cluster discovery, just like the Kafka connector.
* **`key`** - Kafka cluster API key. This is the key portion of a Confluent Cloud API key pair, not the Schema Registry key. Supports `${ENV_VAR}` substitution so you can avoid storing credentials in config files. Internally, e6 Ingestion Engine maps this to the SASL `username`.
* **`secret`** - Kafka cluster API secret. The secret portion of the same API key pair. Supports `${ENV_VAR}` substitution. Internally, e6 Ingestion Engine maps this to the SASL `password`.

When e6 Ingestion Engine builds the underlying Kafka client, it converts the Confluent connection into a SASL config with `protocol: SASL_SSL` and `mechanism: PLAIN`. You never need to set these yourself - the connector does it automatically.

### Schema Registry

If your Confluent Cloud topics use Avro or Protobuf schemas managed by the Confluent Schema Registry, add the schema registry block to the connection. This is optional - skip it if your topics use JSON or raw bytes.

```yaml
schema_registry:
  endpoint: https://psrc-xxxxx.us-east-1.aws.confluent.cloud
  api_key: "${SR_API_KEY}"
  api_secret: "${SR_API_SECRET}"
```

* **`endpoint`** - Schema Registry URL. Find this in your Confluent Cloud environment under **Schema Registry > API endpoint**. When present, e6 Ingestion Engine initializes a schema registry client that fetches Avro and Protobuf schemas for deserialization. If `endpoint` is `null` or missing, the entire schema registry block is ignored.
* **`api_key`** (optional) - Schema Registry API key. This is a **separate** key pair from the cluster API key - Confluent Cloud uses different credentials for the Schema Registry. Supports `${ENV_VAR}` substitution. Serialized only when present (`serde(skip_serializing_if = "Option::is_none")`).
* **`api_secret`** (optional) - Schema Registry API secret. Supports `${ENV_VAR}` substitution.

### Getting Credentials

1. Go to your Confluent Cloud cluster.
2. Click **API Keys** then **Create Key** to generate a cluster API key pair. Copy the key and secret - this is what goes in `key` and `secret`.
3. For Schema Registry, navigate to **Schema Registry** in your environment, then **API credentials** to create a separate API key pair. These go in `schema_registry.api_key` and `schema_registry.api_secret`.

### What the Connector Does Not Expose

Unlike the Kafka connector, the Confluent connector does not expose `authentication` (it's always SASL/PLAIN over SSL) or `connection_properties` at the connection level. The connection starts with an empty `connection_properties` map. If you need to pass extra rdkafka properties, use `client_configs` at the table level - this is supported because the Confluent connector delegates to the Kafka connector for all source and sink operations.

***

## Reading from Confluent Cloud

Source table configuration is identical to the Kafka connector. Configure a source with the topic name, a schema describing the message structure, and an offset mode.

```yaml
topic: user-events
type: source
source:
  offset: latest
schema:
  format:
    json: {}
  fields:
    - field_name: user_id
      field_type:
        type:
          primitive: Int64
      nullable: false
    - field_name: event_type
      field_type:
        type:
          primitive: Utf8
      nullable: false
```

All Kafka source options are available: `offset` (`latest`, `earliest`, `group`), `read_mode` (`read_uncommitted`, `read_committed`), `group_id`, `group_id_prefix`, `value_subject`, and `client_configs`. See [Reading from Kafka](/ingestion-engine/connectors/message-queues/kafka.md#reading-from-kafka) for full details on each option, including how e6 Ingestion Engine handles offset tracking, consumer groups, metadata fields, and schema registry subjects.

***

## Writing to Confluent Cloud

Sink table configuration is identical to the Kafka connector. Configure a sink with the topic name, a delivery guarantee, and a schema describing the output messages.

```yaml
topic: processed-events
type: sink
sink:
  commit_mode: at_least_once
schema:
  format:
    json: {}
  fields:
    - field_name: user_id
      field_type:
        type:
          primitive: Utf8
      nullable: false
    - field_name: event_type
      field_type:
        type:
          primitive: Utf8
      nullable: false
```

All Kafka sink options are available: `commit_mode` (`at_least_once`, `exactly_once`), `key_field`, `timestamp_field`, `value_subject`, and `client_configs`. See [Writing to Kafka](/ingestion-engine/connectors/message-queues/kafka.md#writing-to-kafka) for full details on delivery guarantees, message keys, timestamps, and extra producer properties.

***

## Complete Example

A full configuration reading JSON events from Confluent Cloud and writing processed results back to a different topic.

**Connection:**

```yaml
bootstrap_servers: pkc-xxxxx.us-east-1.aws.confluent.cloud:9092
key: "${CONFLUENT_API_KEY}"
secret: "${CONFLUENT_API_SECRET}"
schema_registry:
  endpoint: https://psrc-xxxxx.us-east-1.aws.confluent.cloud
  api_key: "${SR_API_KEY}"
  api_secret: "${SR_API_SECRET}"
```

**Source** - read user events, starting from latest, with transactional isolation:

```yaml
topic: user-events
type: source
source:
  offset: latest
  read_mode: read_committed
  group_id_prefix: prod-
schema:
  format:
    json: {}
  fields:
    - field_name: user_id
      field_type:
        type:
          primitive: Int64
      nullable: false
    - field_name: event_type
      field_type:
        type:
          primitive: Utf8
      nullable: false
    - field_name: timestamp
      field_type:
        type:
          primitive: TimestampNanosecond
      nullable: false
    - field_name: payload
      field_type:
        type:
          primitive: Utf8
      nullable: true
```

**Sink** - write processed events with exactly-once delivery, keyed by user:

```yaml
topic: processed-events
type: sink
sink:
  commit_mode: exactly_once
  key_field: user_id
  timestamp_field: timestamp
schema:
  format:
    json: {}
  fields:
    - field_name: user_id
      field_type:
        type:
          primitive: Utf8
      nullable: false
    - field_name: event_type
      field_type:
        type:
          primitive: Utf8
      nullable: false
    - field_name: timestamp
      field_type:
        type:
          primitive: TimestampNanosecond
      nullable: false
```

***

## JSON Schema Reference

<details>

<summary>Connection Schema</summary>

```json
{
  "type": "object",
  "required": ["bootstrap_servers", "key", "secret"],
  "properties": {
    "bootstrap_servers": {
      "type": "string",
      "description": "Confluent Cloud bootstrap server address"
    },
    "key": {
      "type": "string",
      "description": "Confluent Cloud cluster API key (maps to SASL username)"
    },
    "secret": {
      "type": "string",
      "description": "Confluent Cloud cluster API secret (maps to SASL password)"
    },
    "schema_registry": {
      "type": "object",
      "description": "Optional. Omit entirely if not using Schema Registry.",
      "properties": {
        "endpoint": {
          "type": ["string", "null"],
          "description": "Schema Registry URL. If null, schema registry is disabled."
        },
        "api_key": {
          "type": "string",
          "description": "Schema Registry API key (separate from cluster key)"
        },
        "api_secret": {
          "type": "string",
          "description": "Schema Registry API secret"
        }
      }
    }
  }
}
```

</details>

<details>

<summary>Table Schema</summary>

Table configuration is identical to the Kafka connector. See [Kafka Table Schema](/ingestion-engine/connectors/message-queues/kafka.md#json-schema-reference) for the full JSON schema.

```json
{
  "type": "object",
  "required": ["topic", "type"],
  "properties": {
    "topic": {"type": "string"},
    "type": {"type": "string", "enum": ["source", "sink"]},
    "source": {
      "type": "object",
      "description": "Source-specific config (when type=source)",
      "properties": {
        "offset": {"type": "string", "enum": ["latest", "earliest", "group"]},
        "read_mode": {"type": "string", "enum": ["read_uncommitted", "read_committed"]},
        "group_id": {"type": "string"},
        "group_id_prefix": {"type": "string"}
      },
      "required": ["offset"]
    },
    "sink": {
      "type": "object",
      "description": "Sink-specific config (when type=sink)",
      "properties": {
        "commit_mode": {"type": "string", "enum": ["at_least_once", "exactly_once"]},
        "key_field": {"type": "string"},
        "timestamp_field": {"type": "string"}
      },
      "required": ["commit_mode"]
    },
    "client_configs": {"type": "object"},
    "value_subject": {"type": "string"}
  }
}
```

</details>


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://docs.e6data.com/ingestion-engine/connectors/message-queues/confluent.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.