Aiven Kafka

Aiven Kafka Message Queue Managed Service

Synopsis

Creates a target for Aiven for Apache Kafka. This target uses the standard Kafka implementation configured for Aiven's managed infrastructure, including required TLS encryption and SASL authentication.

Schema

- name: <string>
  description: <string>
  type: aivenkafka
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    address: <string>
    port: <numeric>
    topic: <string>
    client_id: <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 aivenkafka
pipelines	N	-	Optional post-processor pipelines
status	N	true	Enable/disable the target

Aiven Kafka Connection

Field	Required	Default	Description
address	Y	-	Aiven Kafka service hostname (from Aiven Console). Supports a single address or a comma-separated list for multiple seed brokers. Each entry may include a port; if omitted, the port field value is appended.
port	Y	-	Aiven Kafka service port (typically 12835 or 25835)
client_id	N	-	Client identifier for connection tracking
topic	Y	-	Kafka topic name for message delivery

Aiven Authentication

Aiven for Apache Kafka requires SASL authentication:

Field	Required	Default	Description
algorithm	Y		Must be scram-sha-256 or scram-sha-512 for Aiven
username	Y	-	Aiven Kafka service user (from Aiven Console)
password	Y	-	Aiven Kafka service user password (from Aiven Console)

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

Aiven TLS Configuration

Aiven for Apache Kafka requires TLS encryption:

Field	Required	Default	Description
tls.status	Y		Must be true for Aiven
tls.insecure_skip_verify	N	false	Skip TLS certificate verification (not recommended for production)
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*	-	Client certificate filename if using mutual TLS
tls.key_name	N*	-	Client private key filename if using mutual TLS
tls.passphrase	N*	-	Passphrase for encrypted private key

* = Conditionally required when using mutual TLS authentication.

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

Aiven for Apache Kafka is a fully managed Kafka service that handles infrastructure management, updates, and monitoring. The service requires TLS encryption and SASL authentication for all connections.

DataStream connects to Aiven Kafka using the standard Kafka target type with Aiven-specific configuration. Messages are batched according to configured limits and sent using Kafka's producer API with guaranteed delivery semantics.

Aiven-Specific Requirements

1. TLS Required: All connections must use TLS encryption (tls.status: true)
2. SASL Required: Authentication must use SCRAM-SHA-256 or SCRAM-SHA-512
3. Service URI Format: Use the connection details from Aiven Console
4. Topic Management: Create topics in Aiven Console or enable auto-creation

Getting Aiven Connection Details

From the Aiven Console:

1. Navigate to your Kafka service
2. Find "Connection Information" section
3. Copy the service hostname and port
4. Note the SASL authentication mechanism (SCRAM-SHA-256 or SCRAM-SHA-512)
5. Create or retrieve service user credentials

note

Aiven Kafka services use custom ports (typically 12835 or 25835) instead of the standard Kafka port 9092. Always verify the port from your Aiven Console.

warning

Store Aiven credentials securely. Use environment variables or secret management systems instead of hardcoding credentials in configuration files.

Examples

Basic Aiven Kafka

Minimum configuration for Aiven for Apache Kafka:

targets:
  - name: aiven_kafka
    type: aivenkafka
    properties:
      address: "kafka-project-name.aivencloud.com"
      port: 12835
      topic: "logs.system"
      algorithm: "scram-sha-256"
      username: "avnadmin"
      password: "${AIVEN_KAFKA_PASSWORD}"
      tls:
        status: true

With Compression

Configuration using compression for bandwidth efficiency:

targets:
  - name: aiven_kafka_compressed
    type: aivenkafka
    properties:
      address: "kafka-project-name.aivencloud.com"
      port: 12835
      topic: "logs.application"
      algorithm: "scram-sha-512"
      username: "avnadmin"
      password: "${AIVEN_KAFKA_PASSWORD}"
      compression: "lz4"
      tls:
        status: true
        min_tls_version: "tls1.2"

High Throughput

Configuration optimized for high-volume log delivery:

targets:
  - name: aiven_kafka_highvolume
    type: aivenkafka
    properties:
      address: "kafka-project-name.aivencloud.com"
      port: 25835
      topic: "logs.highvolume"
      algorithm: "scram-sha-256"
      username: "avnadmin"
      password: "${AIVEN_KAFKA_PASSWORD}"
      max_events: 5000
      max_bytes: 10485760
      compression: "zstd"
      acknowledgments: "leader"
      tls:
        status: true

With Auto Topic Creation

Configuration allowing automatic topic creation:

targets:
  - name: aiven_kafka_auto
    type: aivenkafka
    properties:
      address: "kafka-project-name.aivencloud.com"
      port: 12835
      topic: "logs.dynamic"
      algorithm: "scram-sha-256"
      username: "avnadmin"
      password: "${AIVEN_KAFKA_PASSWORD}"
      allow_auto_topic_creation: true
      tls:
        status: true

With Normalization

Configuration using field normalization:

targets:
  - name: aiven_kafka_ecs
    type: aivenkafka
    properties:
      address: "kafka-project-name.aivencloud.com"
      port: 12835
      topic: "logs.ecs"
      algorithm: "scram-sha-256"
      username: "avnadmin"
      password: "${AIVEN_KAFKA_PASSWORD}"
      field_format: "ecs"
      tls:
        status: true

With Pipeline

Using a pipeline for additional log processing:

targets:
  - name: aiven_kafka_pipeline
    type: aivenkafka
    pipelines:
      - enrich_logs
    properties:
      address: "kafka-project-name.aivencloud.com"
      port: 12835
      topic: "logs.enriched"
      algorithm: "scram-sha-256"
      username: "avnadmin"
      password: "${AIVEN_KAFKA_PASSWORD}"
      tls:
        status: true

Production Grade

Configuration with all reliability features enabled:

targets:
  - name: aiven_kafka_production
    type: aivenkafka
    properties:
      address: "kafka-project-name.aivencloud.com"
      port: 12835
      topic: "logs.production"
      client_id: "datastream-prod-001"
      algorithm: "scram-sha-512"
      username: "avnadmin"
      password: "${AIVEN_KAFKA_PASSWORD}"
      compression: "lz4"
      acknowledgments: "all"
      disable_idempotent_write: false
      max_events: 1000
      tls:
        status: true
        min_tls_version: "tls1.3"

Troubleshooting

The aivenkafka 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 Aiven for Apache Kafka

There are no per-broker defaults. algorithm starts at none and tls.status at false, whatever the target type is, and port has to be set for this service. A target built from the Aiven connection details alone is therefore refused at startup with is SASL missing? or is TLS missing?. Set all three yourself.

• algorithm must be scram-sha-256 or scram-sha-512, matching the mechanism your service publishes. Aiven offers SCRAM, so plain is refused with UNSUPPORTED_SASL_MECHANISM even when the credential is correct.
• username and password are the service user and its password from the Aiven Console. A user that was never created, or a password that was reset, gives SASL_AUTHENTICATION_FAILED.
• tls.status must be true. Aiven accepts encrypted connections only.
• port must be set. It is the only one of the three whose default is never right here.

port: 12835
algorithm: "scram-sha-512"
username: "avnadmin"
password: "$secret{id=12}"
tls:
  status: true

TOPIC_AUTHORIZATION_FAILED on every batch means the service user has no write right on the topic named in topic. Grant it in the Aiven Console, which typically issues that right as an ACL entry on the service user scoped to the topic.

The port is never 9092

Leave port out and the target connects to 9092, where an Aiven service has nothing listening. An Aiven service uses a port of its own, typically 12835 or 25835, and it is not the same for every service.

That mistake does not look like an authentication problem. It shows up as a connection failure at startup, reported as i/o timeout, context deadline exceeded, or connection refused, with no mention of SASL or TLS.

tip

Take the port from the Connection Information of your service in the Aiven Console, together with the host name and the mechanism. Alternatively, append it to each entry in address, as in kafka-project-name.aivencloud.com:12835, which overrides port for that entry.