Confluent Cloud

Confluent Message Queue

Synopsis

Creates a target that writes log messages to Confluent Cloud topics with support for batching, compression, and authentication. The target handles message delivery efficiently with configurable batch limits based on size or event count.

Schema

- name: <string>
  description: <string>
  type: confluentcloud
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    address: <string>
    port: <numeric>
    client_id: <string>
    topic: <string>
    algorithm: <string>
    username: <string>
    password: <string>
    compression: <string>
    compression_level: <string>
    acknowledgments: <string>
    allow_auto_topic_creation: <boolean>
    disable_idempotent_write: <boolean>
    max_bytes: <numeric>
    max_events: <numeric>
    batch_mode: <string>
    batch_separator: <string>
    field_format: <string>
    tls:
      status: <boolean>
      insecure_skip_verify: <boolean>
      min_tls_version: <string>
      max_tls_version: <string>
      cert_name: <string>
      key_name: <string>
      passphrase: <string>
    debug:
      status: <boolean>
      dont_send_logs: <boolean>

Configuration

The following fields are used to define the target:

Field	Required	Default	Description
name	Y		Target name
description	N	-	Optional description
type	Y		Must be confluentcloud
pipelines	N	-	Optional post-processor pipelines
status	N	true	Enable/disable the target

Confluent Cloud Connection

Field	Required	Default	Description
address	Y	-	Confluent Cloud bootstrap server. Supports a single address or a comma-separated list for multiple seed brokers (e.g., "pkc-1.confluent.cloud:9092,pkc-2.confluent.cloud:9092"). Each entry may include a port; if omitted, the port field value is appended.
port	N	9092	Confluent Cloud broker port
client_id	N	-	Client identifier for connection tracking
topic	Y	-	Kafka topic name for message delivery

Authentication

Field	Required	Default	Description
algorithm	Y	"none"	Authentication mechanism. Confluent Cloud requires plain, so set it explicitly — the default does not authenticate
username	Y	-	Confluent Cloud API Key
password	Y	-	Confluent Cloud API Secret

note

Confluent Cloud uses SASL/PLAIN authentication. Use your Confluent Cloud API Key as the username and API Secret as the password.

Producer Settings

Field	Required	Default	Description
compression	N	"none"	Message compression: none, gzip, snappy, lz4, zstd
compression_level	N	-	Compression level (algorithm-specific)
acknowledgments	N	"leader"	Acknowledgment level: none, leader, all. none and leader are silently upgraded to all unless disable_idempotent_write: true is also set — the idempotent producer is on by default and requires all
allow_auto_topic_creation	N	false	Allow automatic topic creation if topic doesn't exist
disable_idempotent_write	N	false	Turn off the idempotent producer. Required if you want acknowledgments to actually take effect as leader or none; leaving it false keeps exactly-once-per-partition write semantics

Batch Configuration

Field	Required	Default	Description
max_bytes	N	1048576	Maximum batch size in bytes (max: 104857600). 0 does not remove the ceiling: the largest message the producer sends is then 1000012 bytes, the Kafka default maximum message size
max_events	N	1000	Maximum number of events per batch
batch_mode	N	individual	Output format: individual (one message per event), json_array (array of JSON objects), jsonl (JSON Lines with separator)
batch_separator	N	,	Separator between messages when using jsonl batch mode
field_format	N	-	Data normalization format. See applicable Normalization section

note

Batches are sent when either max_bytes or max_events limit is reached, whichever comes first.

TLS Configuration

Field	Required	Default	Description
tls.status	N	false	Enable TLS/SSL encryption. Confluent Cloud accepts TLS connections only, so set it to true explicitly
tls.insecure_skip_verify	N	false	Skip certificate verification (not recommended)
tls.min_tls_version	N	tls1.2	Minimum TLS version: tls1.0, tls1.1, tls1.2, tls1.3
tls.max_tls_version	N	tls1.3	Maximum TLS version: tls1.0, tls1.1, tls1.2, tls1.3
tls.ca_name	N	-	CA bundle used to verify the broker certificate. When unset the host trust store is used; when set it replaces it
tls.cert_name	N	"cert.pem"	Client certificate file name for mTLS
tls.key_name	N	"key.pem"	Private key file name for mTLS
tls.passphrase	N	-	Passphrase for encrypted private key

note

Confluent Cloud requires TLS encryption. The tls.status field should always be true.

Scheduling

See Scheduling and Pool Behavior for interval and cron fields shared by all targets.

Debug Options

Field	Required	Default	Description
debug.status	N	false	Enable debug logging
debug.dont_send_logs	N	false	Process logs but don't send to target (testing)

Details

Confluent Cloud is a fully managed Apache Kafka service. This target type allows you to connect to Confluent Cloud clusters using the standard Kafka protocol with SASL/PLAIN authentication.

Authentication

Confluent Cloud uses API Keys for authentication. When creating the target:

• Set algorithm to "plain"
• Use your Confluent Cloud API Key as the username
• Use your Confluent Cloud API Secret as the password
• Enable TLS with tls.status: true

API Keys can be created in the Confluent Cloud Console under your cluster settings.

Connection Requirements

• TLS must be enabled (Confluent Cloud requires encrypted connections)
• SASL/PLAIN authentication is required
• Bootstrap server address format: pkc-xxxxx.region.cloud-provider.confluent.cloud
• Port is typically 9092

Message Delivery Guarantees

The acknowledgments setting controls delivery guarantees:

Level	Behavior	Use Case
none	No acknowledgment from broker	Maximum throughput, lowest durability
leader	Acknowledgment from partition leader only	Balanced throughput and durability
all	Acknowledgment from all in-sync replicas	Maximum durability (recommended)

Compression

Message compression reduces network bandwidth and storage costs:

Algorithm	Compression Ratio	CPU Usage	Speed
none	None	Minimal	Fastest
gzip	High	High	Slow
snappy	Medium	Low	Fast
lz4	Medium	Low	Fast
zstd	High	Medium	Medium

Examples

Basic Configuration

The minimum configuration for a Confluent Cloud target:

targets:
  - name: basic_confluent
    type: confluentcloud
    properties:
      address: "pkc-xxxxx.us-east-1.aws.confluent.cloud"
      topic: "application-logs"
      algorithm: "plain"
      username: "ABCDEFGHIJ123456"
      password: "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ12"
      tls:
        status: true

With Compression

Configuration with zstd compression:

targets:
  - name: compressed_confluent
    type: confluentcloud
    properties:
      address: "pkc-xxxxx.us-east-1.aws.confluent.cloud"
      topic: "application-logs"
      algorithm: "plain"
      username: "ABCDEFGHIJ123456"
      password: "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ12"
      compression: "zstd"
      tls:
        status: true

High Throughput

Configuration optimized for maximum throughput:

targets:
  - name: high_throughput_confluent
    type: confluentcloud
    properties:
      address: "pkc-xxxxx.us-east-1.aws.confluent.cloud"
      topic: "high-volume-logs"
      algorithm: "plain"
      username: "ABCDEFGHIJ123456"
      password: "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ12"
      compression: "lz4"
      acknowledgments: "leader"
      max_bytes: 1048576
      max_events: 10000
      tls:
        status: true

High Reliability

Configuration optimized for maximum durability:

targets:
  - name: reliable_confluent
    type: confluentcloud
    pipelines:
      - checkpoint
    properties:
      address: "pkc-xxxxx.us-east-1.aws.confluent.cloud"
      topic: "critical-logs"
      algorithm: "plain"
      username: "ABCDEFGHIJ123456"
      password: "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ12"
      compression: "zstd"
      acknowledgments: "all"
      max_events: 100
      disable_idempotent_write: false
      tls:
        status: true
        min_tls_version: "tls1.3"

With Field Normalization

Using field normalization for standard format:

targets:
  - name: normalized_confluent
    type: confluentcloud
    properties:
      address: "pkc-xxxxx.us-east-1.aws.confluent.cloud"
      topic: "normalized-logs"
      algorithm: "plain"
      username: "ABCDEFGHIJ123456"
      password: "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ12"
      field_format: "cim"
      compression: "snappy"
      tls:
        status: true

Azure Region

Configuration for Confluent Cloud on Azure:

targets:
  - name: confluent_azure
    type: confluentcloud
    properties:
      address: "pkc-yyyyy.eastus.azure.confluent.cloud"
      topic: "azure-logs"
      algorithm: "plain"
      username: "ABCDEFGHIJ123456"
      password: "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ12"
      compression: "zstd"
      acknowledgments: "all"
      tls:
        status: true

GCP Region

Configuration for Confluent Cloud on GCP:

targets:
  - name: confluent_gcp
    type: confluentcloud
    properties:
      address: "pkc-zzzzz.us-central1.gcp.confluent.cloud"
      topic: "gcp-logs"
      algorithm: "plain"
      username: "ABCDEFGHIJ123456"
      password: "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ12"
      compression: "zstd"
      acknowledgments: "all"
      tls:
        status: true

With Client ID

Configuration with client ID for tracking:

targets:
  - name: confluent_tracked
    type: confluentcloud
    properties:
      address: "pkc-xxxxx.us-east-1.aws.confluent.cloud"
      topic: "tracked-logs"
      client_id: "datastream-producer-01"
      algorithm: "plain"
      username: "ABCDEFGHIJ123456"
      password: "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ12"
      compression: "lz4"
      tls:
        status: true

Scheduled Batching

Configuration with scheduled batch delivery:

targets:
  - name: scheduled_confluent
    type: confluentcloud
    properties:
      address: "pkc-xxxxx.us-east-1.aws.confluent.cloud"
      topic: "scheduled-logs"
      algorithm: "plain"
      username: "ABCDEFGHIJ123456"
      password: "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ12"
      max_events: 5000
      interval: "5m"
      compression: "gzip"
      tls:
        status: true

Auto Topic Creation

Configuration with automatic topic creation enabled:

targets:
  - name: auto_topic_confluent
    type: confluentcloud
    properties:
      address: "pkc-xxxxx.us-east-1.aws.confluent.cloud"
      topic: "dynamic-logs"
      algorithm: "plain"
      username: "ABCDEFGHIJ123456"
      password: "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ12"
      allow_auto_topic_creation: true
      compression: "snappy"
      max_events: 500
      tls:
        status: true

Troubleshooting

The confluentcloud target behaves the same way as the kafka target. It opens the connection the same way, produces the same batches, and reports the same errors for the same causes. Use the Apache Kafka troubleshooting section for the full list of errors, causes and fixes. See Target Delivery Errors for how Director logs and retries target failures.

What differs for Confluent Cloud

There are no per-broker defaults. algorithm starts at none, port at 9092 and tls.status at false, whatever the target type is. A target built from the Confluent Cloud connection details alone is therefore refused at startup with is SASL missing? or is TLS missing?. Set all three yourself.

• algorithm must be plain. Confluent Cloud offers SASL/PLAIN, so scram-sha-256 and scram-sha-512 are refused with UNSUPPORTED_SASL_MECHANISM even when the credential is correct.
• username is the API Key and password is the API Secret, both from the Confluent Cloud Console. A key that was deleted or rotated gives SASL_AUTHENTICATION_FAILED.
• tls.status must be true. Confluent Cloud accepts encrypted connections only.
• port is 9092, which is also the default, so it is the one of the three you can usually leave alone.

port: 9092
algorithm: "plain"
username: "ABCDEFGHIJ123456"
password: "$secret{id=12}"
tls:
  status: true

TOPIC_AUTHORIZATION_FAILED on every batch means the API Key carries no write right on the topic named in topic. Grant it in the Confluent Cloud Console, which typically issues that right as an ACL or a role binding scoped to the topic.

Topics have to exist before you send

allow_auto_topic_creation is passed to the cluster, but a managed cluster typically refuses topic creation by a client. Setting it to true therefore does not save you from UNKNOWN_TOPIC_OR_PARTITION on every batch.

tip

Create the topic in the Confluent Cloud Console first, then check that topic matches its name exactly, including case and any environment prefix. Startup cannot catch a wrong name, because the check it makes does not mention your topic.