Confluent

Source | Sink The Confluent connector is a convenience wrapper around the Kafka connector. 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, Ingest 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 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 - Ingest 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

Ingest 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. Ingest 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 Ingest 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 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

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

Azure
AWS
GCP

Kafka bootstrap: outbound TCP port 9092 (SASL_SSL) from the Ingest 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.

Kafka bootstrap: outbound TCP port 9092 (SASL_SSL) from the Ingest 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

Kafka bootstrap: outbound TCP port 9092 (SASL_SSL) from the Ingest 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

Secret Rotation

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

Create a new key in Confluent Cloud
Update the key in the Ingest Engine connection config
Confirm the pipeline reconnects successfully with the new key
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.

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. Ingest 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, Ingest 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, Ingest Engine maps this to the SASL password.

When Ingest 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.

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, Ingest 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

Go to your Confluent Cloud cluster.
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.
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.

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 for full details on each option, including how Ingest 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.

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 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:

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:

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:

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

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Prerequisites

Confluent Cloud Resources

Credentials

Service Account ACLs

Network Access

Secret Rotation

Connection

Schema Registry

Getting Credentials

What the Connector Does Not Expose

Reading from Confluent Cloud

Writing to Confluent Cloud

Complete Example

JSON Schema Reference

​Prerequisites

​Confluent Cloud Resources

​Credentials

​Service Account ACLs

​Network Access

​Secret Rotation

​Connection

​Schema Registry

​Getting Credentials

​What the Connector Does Not Expose

​Reading from Confluent Cloud

​Writing to Confluent Cloud

​Complete Example

​JSON Schema Reference

Prerequisites

Confluent Cloud Resources

Credentials

Service Account ACLs

Network Access

Secret Rotation

Connection

Schema Registry

Getting Credentials

What the Connector Does Not Expose

Reading from Confluent Cloud

Writing to Confluent Cloud

Complete Example

JSON Schema Reference