Apache Kafka

Apache Kafka Message Queue

Synopsis

Creates a target that writes log messages to Apache Kafka topics with support for batching, compression, and various authentication methods. The target handles message delivery efficiently with configurable batch limits based on size or event count. Apache Kafka is an open-source distributed event streaming platform for high-throughput, fault-tolerant data pipelines.

Schema

- name: <string>
  description: <string>
  type: kafka
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    address: <string>
    port: <numeric>
    client_id: <string>
    topic: <string>
    algorithm: <string>
    username: <string>
    password: <string>
    kerberos_keytab: <string>
    kerberos_principal: <string>
    kerberos_realm: <string>
    kerberos_kdc: <string>
    kerberos_config: <string>
    kerberos_service_name: <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>
      ca_name: <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 kafka
pipelines	N	-	Optional post-processor pipelines
status	N	true	Enable/disable the target

Kafka Connection

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

Authentication

Field	Required	Default	Description
algorithm	N	"none"	Authentication mechanism: none, plain, scram-sha-256, scram-sha-512, gssapi
username	N*	-	Username for SASL authentication
password	N*	-	Password for SASL authentication
kerberos_keytab	N**	-	Keytab file path, or base64-encoded keytab content. Resolvable via ${ENV_VAR} or $secret{...}
kerberos_principal	N	username	Client principal (user or user@REALM). Falls back to username when omitted
kerberos_realm	N	-	Kerberos realm. Overrides the realm derived from the principal or krb5.conf
kerberos_kdc	N**	-	Comma-separated KDC addresses (host[:port], default port 88)
kerberos_config	N**	-	krb5.conf file path, or inline krb5.conf content. Generated from kerberos_realm and kerberos_kdc when omitted
kerberos_service_name	N	"kafka"	Kafka broker service principal name (SPN)

* = Conditionally required when algorithm is not none.

** = Applies only when algorithm is gssapi. kerberos_keytab is required, along with either kerberos_kdc or kerberos_config.

Producer Settings

Field	Required	Default	Description
compression	N	"none"	Message compression: none, gzip, snappy, lz4, zstd
compression_level	N	-	Compression level (algorithm-specific, see details below)
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. Values between 1 and 511 are refused
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
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. The named authority is trusted in addition to the host trust store
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

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

Apache Kafka is an open-source distributed event streaming platform. This target allows you to send log messages to Kafka topics for real-time processing and analytics.

Authentication Methods

Kafka supports multiple SASL authentication mechanisms:

No Authentication

• Set algorithm: "none"
• No username or password required
• Suitable for development or internal networks

SASL/PLAIN

• Simple username/password authentication
• Set algorithm: "plain"
• Credentials sent over the connection (use TLS for security)

SASL/SCRAM-SHA-256

• Secure challenge-response authentication
• Set algorithm: "scram-sha-256"
• More secure than PLAIN

SASL/SCRAM-SHA-512

• Enhanced secure authentication with SHA-512
• Set algorithm: "scram-sha-512"
• Recommended for production use

SASL/GSSAPI (Kerberos)

• Kerberos authentication for enterprise environments with a KDC
• Set algorithm: "gssapi" (kerberos is also accepted)
• Authenticates with a keytab specified by kerberos_keytab (file path or base64-encoded content)
• The client principal is taken from kerberos_principal, falling back to username when omitted
• Provide the KDC via kerberos_kdc, or supply a full krb5.conf via kerberos_config; when only kerberos_kdc and kerberos_realm are set, a minimal krb5.conf is generated automatically
• kerberos_service_name sets the broker SPN (default kafka)

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, lower throughput

Idempotent Producer

By default, the target uses idempotent writes to prevent message duplication. This ensures exactly-once semantics within a single producer session. The disable_idempotent_write option allows disabling this feature, but it is not recommended for production use.

Compression

Message compression reduces network bandwidth and storage requirements:

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

Compression Levels

Optional compression levels can be specified with the compression_level parameter:

LZ4 Compression Levels:

• fast (default)
• level1 through level9 (increasing compression ratio)

Gzip Compression Levels:

• nocompression
• bestspeed
• defaultcompression (default)
• bestcompression
• huffmanonly

Zstd Compression Levels:

• speedfastest
• speeddefault (default)
• speedbettercompression
• speedbestcompression

Batch Processing

The target accumulates messages in memory and sends them in batches. Batches are sent when either the byte limit (max_bytes) or event count limit (max_events) is reached.

Setting max_bytes to 0 removes the local ceiling but not the one on the wire: the producer still refuses to send a message larger than 1000012 bytes, the Kafka default maximum message size. This matters in the json_array and jsonl batch modes, where the whole batch travels as a single Kafka message, so max_events events have to fit within that limit. In individual mode each event is its own message and only a single oversized event is affected.

Topic Management

Kafka topics must exist before sending messages unless allow_auto_topic_creation is enabled. When enabled, topics are created with broker default settings, but only if the broker itself lets clients create them, which managed services typically do not. For production environments, create topics in advance with appropriate partition counts and replication factors.

TLS/SSL Security

The target supports encrypted connections using TLS/SSL. For production environments:

• Set tls.status: true to enable encryption
• Set tls.insecure_skip_verify: false to validate certificates
• Use TLS 1.2 or higher for security compliance
• Configure mTLS with client certificates for mutual authentication

Examples

Basic Configuration

The minimum configuration for a Kafka target:

targets:
  - name: basic_kafka
    type: kafka
    properties:
      address: "kafka.example.com"
      topic: "application-logs"

With SASL/PLAIN Authentication

Configuration using PLAIN authentication:

targets:
  - name: authenticated_kafka
    type: kafka
    properties:
      address: "kafka.example.com"
      port: 9092
      topic: "secure-logs"
      algorithm: "plain"
      username: "log-producer"
      password: "secure-password"

With SASL/SCRAM-SHA-256

Configuration using SCRAM-SHA-256 authentication:

targets:
  - name: scram_kafka
    type: kafka
    properties:
      address: "kafka.example.com"
      port: 9093
      topic: "authenticated-logs"
      algorithm: "scram-sha-256"
      username: "kafka-producer"
      password: "secure-password-here"
      tls:
        status: true

With SASL/SCRAM-SHA-512

Configuration using SCRAM-SHA-512 authentication:

targets:
  - name: secure_kafka
    type: kafka
    properties:
      address: "kafka.example.com"
      port: 9093
      topic: "production-logs"
      algorithm: "scram-sha-512"
      username: "producer-user"
      password: "strong-password"
      tls:
        status: true
        insecure_skip_verify: false

With SASL/GSSAPI (Kerberos)

Configuration using Kerberos authentication with a keytab and KDC:

targets:
  - name: kerberos_kafka
    type: kafka
    properties:
      address: "kafka.example.com"
      port: 9093
      topic: "secured-logs"
      algorithm: "gssapi"
      kerberos_keytab: "$secret{id=12}"
      kerberos_principal: "vmetric@EXAMPLE.COM"
      kerberos_kdc: "kdc01.example.com,kdc02.example.com:88"
      kerberos_service_name: "kafka"
      tls:
        status: true

With TLS Encryption

Configuration with TLS enabled:

targets:
  - name: encrypted_kafka
    type: kafka
    properties:
      address: "kafka.example.com"
      port: 9093
      topic: "encrypted-logs"
      tls:
        status: true
        insecure_skip_verify: false
        min_tls_version: "tls1.2"
        max_tls_version: "tls1.3"

With mTLS Authentication

Configuration using mutual TLS authentication:

targets:
  - name: mtls_kafka
    type: kafka
    properties:
      address: "kafka.example.com"
      port: 9093
      topic: "mtls-logs"
      tls:
        status: true
        insecure_skip_verify: false
        min_tls_version: "tls1.2"
        cert_name: "client-cert.pem"
        key_name: "client-key.pem"
        passphrase: "key-passphrase"

With Compression

Configuration with zstd compression:

targets:
  - name: compressed_kafka
    type: kafka
    properties:
      address: "kafka.example.com"
      topic: "compressed-logs"
      compression: "zstd"
      compression_level: "speedbettercompression"

High Throughput

Configuration optimized for maximum throughput:

targets:
  - name: high_throughput_kafka
    type: kafka
    properties:
      address: "kafka.example.com"
      topic: "high-volume-logs"
      compression: "lz4"
      compression_level: "fast"
      acknowledgments: "leader"
      max_bytes: 1048576
      max_events: 10000

High Reliability

Configuration optimized for maximum durability:

targets:
  - name: reliable_kafka
    type: kafka
    pipelines:
      - checkpoint
    properties:
      address: "kafka.example.com"
      topic: "critical-logs"
      compression: "zstd"
      acknowledgments: "all"
      max_events: 100
      disable_idempotent_write: false
      tls:
        status: true

With Field Normalization

Using field normalization for standard format:

targets:
  - name: normalized_kafka
    type: kafka
    properties:
      address: "kafka.example.com"
      topic: "normalized-logs"
      field_format: "cim"
      compression: "snappy"

Multi-Broker Cluster

Configuration with multiple seed brokers for cluster discovery:

targets:
  - name: cluster_kafka
    type: kafka
    properties:
      address: "kafka-broker-1.example.com:9092,kafka-broker-2.example.com:9092,kafka-broker-3.example.com:9092"
      client_id: "datastream-producer-01"
      topic: "cluster-logs"
      algorithm: "scram-sha-512"
      username: "producer-user"
      password: "cluster-password"
      compression: "zstd"
      acknowledgments: "all"

Auto Topic Creation

Configuration with automatic topic creation:

targets:
  - name: auto_topic_kafka
    type: kafka
    properties:
      address: "kafka.example.com"
      topic: "dynamic-logs"
      allow_auto_topic_creation: true
      compression: "snappy"
      max_events: 500

Scheduled Batching

Configuration with scheduled batch delivery:

targets:
  - name: scheduled_kafka
    type: kafka
    properties:
      address: "kafka.example.com"
      topic: "scheduled-logs"
      max_events: 5000
      interval: "5m"
      compression: "gzip"
      compression_level: "bestcompression"

Development Environment

Configuration for local development without security:

targets:
  - name: dev_kafka
    type: kafka
    properties:
      address: "localhost"
      port: 9092
      topic: "dev-logs"
      algorithm: "none"
      allow_auto_topic_creation: true

Production Environment

Configuration for production with full security:

targets:
  - name: prod_kafka
    type: kafka
    pipelines:
      - checkpoint
    properties:
      address: "kafka-prod.example.com"
      port: 9093
      client_id: "datastream-prod-01"
      topic: "production-logs"
      algorithm: "scram-sha-512"
      username: "prod-kafka-user"
      password: "strong-production-password"
      compression: "zstd"
      compression_level: "speeddefault"
      acknowledgments: "all"
      max_events: 1000
      disable_idempotent_write: false
      tls:
        status: true
        insecure_skip_verify: false
        min_tls_version: "tls1.2"
        max_tls_version: "tls1.3"

Troubleshooting

This section covers the errors you are most likely to see with the kafka target, what causes each one, and how to fix it. The same behavior applies to the other Kafka-compatible brokers documented on their own pages: Confluent Cloud, Aiven Kafka, Redpanda, WarpStream, Amazon MSK, and IBM Event Streams. Where one of those differs, the entry says so.

Where to look:

• Director logs. Target errors are tagged with the target name and carry "Section":"SenderPool". The part after Reason: or after the last colon is the actual cause.
• The target's connection status in the web interface. It shows the same reason as the log line.

See Target Delivery Errors for how Director logs and retries target failures.

Which permission or setting is missing?

When the target starts, it opens a connection to one of the seed brokers and asks for cluster metadata. That proves the address, TLS and the credentials, so those problems appear at startup. It asks nothing about your topic, so the topic name and your right to write to it are first tested when the first batch is produced, usually seconds later. Match the error you see against this table.

Error text	What is missing	Where to fix it
is SASL missing?	Authentication is not configured. algorithm defaults to none on every broker type	algorithm, with username and password
UNSUPPORTED_SASL_MECHANISM	The value of algorithm is not a mechanism this listener offers	algorithm, matched to the mechanism published for that port
SASL_AUTHENTICATION_FAILED	The credential is wrong, revoked, or resolved to an empty value	username and password
is TLS missing?	The listener speaks TLS and the target does not	tls.status
x509: certificate signed by unknown authority	The broker certificate is signed by an authority the host does not trust	tls.ca_name, or the host trust store
remote error: tls: certificate required	The listener asks every client for a certificate	tls.cert_name and tls.key_name
TOPIC_AUTHORIZATION_FAILED	The producing principal has no WRITE right on the topic, which also grants DESCRIBE	The broker, or the provider's own console
UNKNOWN_TOPIC_OR_PARTITION	The topic does not exist, and the target does not create it	topic, or the broker
CLUSTER_AUTHORIZATION_FAILED	The principal may not open an idempotent producer session	IDEMPOTENT_WRITE on the cluster, or disable_idempotent_write

On a managed service the topic right is granted in the provider's own console rather than with a Kafka command, typically as an ACL or a role scoped to the topic: a write ACL or role binding on Confluent Cloud, a service-user ACL on Aiven Kafka, an API key that carries write permission on WarpStream, a writer role on IBM Event Streams, and an ACL on the SCRAM user on Amazon MSK.

note

The mechanism and the port belong together. A broker usually offers one mechanism per listener, so plain sent to a listener that only offers SCRAM is refused with UNSUPPORTED_SASL_MECHANISM even when the credential is correct. Take both the port and the mechanism from the broker's own documentation, and set port explicitly whenever the service does not use 9092.

warning

IAM authentication for Amazon MSK is not supported by this target. Use SASL/SCRAM or mTLS against an MSK cluster instead. A target pointed at the IAM listener fails at startup with is SASL missing? or UNSUPPORTED_SASL_MECHANISM.

"is SASL missing?" when the broker expects authentication

[Error] [director] [target-<target id>] [prod_kafka] Failed to reinitialize target "prod_kafka" (attempt 2). Reason: failed to create kafka producer: failed to ping kafka broker: broker closed the connection immediately after a request was issued, which often happens when SASL is required but not provided: is SASL missing?

Cause: algorithm is still at its default of none, so the target offers no credential and the broker hangs up. The default is none for every broker type, managed services included, so a target built from connection details alone does not authenticate until you set the field. A bare EOF in place of the longer message means the same thing.

Fix: set algorithm to the mechanism your broker offers, then set username and password. Managed services usually pair that with TLS on the same listener, so enable it at the same time.

algorithm: "scram-sha-512"
username: "log-producer"
password: "$secret{id=12}"
tls:
  status: true

Nothing is sent while this lasts, and the target retries until you fix the cause.

"SASL_AUTHENTICATION_FAILED: SASL Authentication failed."

[Error] [director] [target-<target id>] [prod_kafka] Failed to reinitialize target "prod_kafka" (attempt 12). Reason: failed to create kafka producer: failed to ping kafka broker: SASL_AUTHENTICATION_FAILED: SASL Authentication failed.

Cause: the mechanism was accepted but the credential was rejected. Most brokers append their own wording after the message, typically naming an invalid user name or password. The usual reasons are a key that was rotated or deleted, a secret pasted with trailing whitespace, or a user that was never created on the cluster.

Fix: re-issue the credential and paste both halves again. Check that a ${ENV_VAR} or $secret{...} reference resolves to a non-empty value on the Director host: an empty field is refused earlier with PLAIN user and pass must be non-empty or SCRAM-SHA-512 user and pass must be non-empty. When the reference cannot be resolved at all, the reason names the field instead, as in failed to resolve password: credential: store "..." not found in configuration. Nothing is sent while this lasts, and the target retries until you fix the cause.

"UNSUPPORTED_SASL_MECHANISM: The broker does not support the requested SASL mechanism."

[Error] [director] [target-<target id>] [prod_kafka] Failed to reinitialize target "prod_kafka" (attempt 3). Reason: failed to create kafka producer: failed to ping kafka broker: UNSUPPORTED_SASL_MECHANISM: The broker does not support the requested SASL mechanism.

Cause: algorithm names a mechanism this listener does not offer. The credential is never checked, so a correct user name and password do not help. Pointing a SCRAM configuration at a listener that offers only PLAIN, or the reverse, produces this, and so does any SASL configuration aimed at a port where authentication is switched off.

Fix: take the mechanism and the port from the broker, then set algorithm to plain, scram-sha-256, scram-sha-512, or gssapi to match. Underscores are not accepted, so write scram-sha-512 rather than scram_sha_512. Nothing is sent while this lasts, and the target retries until you fix the cause.

"is TLS missing?", "is TLS misconfigured", or a certificate error

[Error] [director] [target-<target id>] [prod_kafka] Failed to reinitialize target "prod_kafka" (attempt 1). Reason: failed to create kafka producer: failed to ping kafka broker: broker closed the connection immediately during api versions negotiation, which often happens when the broker requires TLS but the client is using plaintext: is TLS missing?

Cause: the encryption settings and the listener disagree, or the certificate cannot be verified. The wording tells you which.

• is TLS missing? means the listener requires TLS while tls.status is still false, its default. Every managed service listener requires it.
• tls: first record does not look like a TLS handshake, or is TLS misconfigured on the client or the broker?, is the opposite: tls.status is true and the port is a plaintext listener.
• tls: failed to verify certificate: x509: certificate signed by unknown authority means the broker presents a certificate from a private certificate authority that the Director host does not trust.
• x509: certificate is valid for ..., not ... means address holds an IP address, or a name that is not in the certificate.
• remote error: tls: certificate required or remote error: tls: bad certificate means the listener wants a client certificate that was not sent or was not trusted.

Fix: set tls.status to match the port you connect to. For a private authority, set tls.ca_name to the CA bundle, either as inline PEM or as a file under the service root, and connect by the host name the certificate carries rather than by IP. For a listener that demands client certificates, set both tls.cert_name and tls.key_name, plus tls.passphrase when the key is encrypted.

Two startup failures point at the files themselves. ca_name "..." could not be loaded (inline PEM, or a path under the service root) means the bundle is unreadable or sits outside the service root. invalid tls configuration means exactly one of the certificate and key was loaded, so place both under the service root and check that the service account can read them. Nothing is sent while any of these lasts, and the target retries until you fix the cause.

"TOPIC_AUTHORIZATION_FAILED: Not authorized to access topics"

[Error] [director] [target-<target id>] [prod_kafka] Sender worker 1 execute() failed for prod_kafka: failed to finalize target cache: failed to produce batch to kafka: failed to produce batch: TOPIC_AUTHORIZATION_FAILED: Not authorized to access topics: [Topic authorization failed.]

Cause: the credential is valid for the cluster but carries no producer right on this topic, so the target starts, reports connected, and then fails on every batch. Startup cannot catch this, because the metadata request it makes does not name your topic.

Fix: grant WRITE on the topic in topic to the principal the target authenticates as, using the provider mapping listed above the entries. The right is per topic, so a target that is re-pointed at a new topic needs the grant again. If the same principal can write with another tool, compare the exact topic name, including case and any environment prefix.

No data is lost. The batch is discarded from memory, the payloads stay queued, and they are redelivered until the grant is in place, so expect the queue to grow meanwhile.

"UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition."

[Error] [director] [target-<target id>] [prod_kafka] Sender worker 0 execute() failed for prod_kafka: failed to finalize target cache: failed to produce batch to kafka: failed to produce batch: UNKNOWN_TOPIC_OR_PARTITION: This server does not host this topic-partition.

Cause: the topic named in topic does not exist on the cluster, and allow_auto_topic_creation is false by default. Like the authorization error above, it appears at the first flush rather than at startup. A principal that may write but not describe the topic can also produce it.

Fix: create the topic with the partition count and replication factor you want, or correct topic. Setting allow_auto_topic_creation: true only helps when the broker itself permits clients to create topics, which managed services typically do not.

A second message belongs here. If topic is left empty, the target still starts and every flush fails with cannot produce record with no topic and no default topic. Set topic, and note that it is not checked before the target starts. In both cases nothing is lost, and the batch is redelivered until the topic is reachable.

"failed to ping kafka broker" with a connection or DNS error

[Error] [director] [target-<target id>] [prod_kafka] Failed to reinitialize target "prod_kafka" (attempt 15). Reason: failed to create kafka producer: failed to ping kafka broker: dial tcp: lookup broker.example.com: no such host

Cause: the target cannot open a connection to any seed broker. The text after the address says which step failed.

• no such host means the name in address does not resolve from the Director host.
• connection refused means nothing is listening on that port. On Windows the wording is connectex: No connection could be made because the target machine actively refused it.
• i/o timeout or context deadline exceeded means the packets are being dropped, usually by a firewall or a security group, or the broker did not answer within the ten seconds allowed for the startup check.

Fix: correct address and port, and open the port from the Director host to every broker. Each entry in address may carry its own port, and entries without one get port appended, so a comma-separated list mixing both forms is fine. Note that the connection is plain TCP and ignores HTTP_PROXY, so a proxy-only egress path does not work. When only one entry is unreachable the error still appears, because the reason reported is the last seed tried. Nothing is sent while this lasts, and the target retries until you fix the cause.

"record rejected by target: record size ... exceeds configured maxBytes ..."

[Error] [director] [target-<target id>] [prod_kafka] Sender worker 2 deterministic failure for prod_kafka after 4 attempts — dropping (giving up): record rejected by target: record size 1258341 exceeds configured maxBytes 1048576

Cause: one event on its own is larger than max_bytes, which defaults to 1048576, so it can never be placed in a batch. The measured size is slightly larger than the event, because the per-record overhead and, in the batched modes, the separator are counted as well. The setting is written max_bytes in the configuration even though the message spells it maxBytes.

Fix: raise max_bytes, keeping it at or below what the broker accepts as a message, or trim the event in a pipeline before it reaches the target. This is the one Kafka error where data is discarded: the payload is dropped after 4 deliveries, and the connection stays up while the rest of the traffic flows normally.

"MESSAGE_TOO_LARGE: The request included a message larger than the max message size the server will accept"

[Error] [director] [target-<target id>] [prod_kafka] Sender worker 1 execute() failed for prod_kafka: failed to finalize target cache: failed to produce JSON array batch to kafka: failed to produce batch: MESSAGE_TOO_LARGE: The request included a message larger than the max message size the server will accept (uncompressed_bytes=1048601)

Cause: the whole batch went over the message limit. With batch_mode set to json_array or jsonl the entire batch travels as a single Kafka message, so max_events events have to fit within one message. Two limits apply: the one the broker enforces, and the one the producer applies locally, which equals max_bytes or, when max_bytes is 0, 1000012 bytes. A batch counted as within max_bytes can still exceed the wire limit once the framing around it is added.

Fix: lower max_events, or lower max_bytes so the batch has headroom below both limits, and raise the broker's own maximum message size if you control it. Switching batch_mode to individual removes the problem, because each event is then its own message.

warning

Nothing is dropped here and nothing is delivered either: the same batch is redelivered indefinitely and fails identically each time, holding up everything queued behind it. Treat it as urgent and change one of the two settings.

"CLUSTER_AUTHORIZATION_FAILED: Cluster authorization failed."

[Error] [director] [target-<target id>] [prod_kafka] Sender worker 1 execute() failed for prod_kafka: failed to finalize target cache: failed to produce batch to kafka: failed to produce batch: unable to initialize a producer ID due to request failures: CLUSTER_AUTHORIZATION_FAILED: Cluster authorization failed.

Cause: the idempotent producer is on by default, and it asks the cluster for a producer ID before the first batch. The principal is allowed to write to the topic but not to make that request.

Fix: grant IDEMPOTENT_WRITE on the cluster to the principal. Older brokers typically require it as a right of its own, separate from the write right on the topic. If you cannot change cluster ACLs, set disable_idempotent_write: true, and note that acknowledgments then takes effect as written instead of being raised to all. Nothing is lost, and the batch is redelivered until one of the two changes is made.

The target stays connected but nothing arrives

Nothing fails here, so there is no error text to search for. Check the following in order.

1. A produce call is waiting on the broker. If a broker or a partition leader becomes unreachable after the target has started, the batch in flight waits for it, with no log line and with the connection status still showing connected. The queue depth is the only symptom. It clears on its own once the broker answers again. The same applies to a partition with too few in-sync replicas, and to a broker that is throttling the producer against a quota, which is retried internally and never logged.

2. The addresses the cluster advertises are unreachable. address is only used to find the cluster. The brokers then answer with their own advertised addresses, and batches go to those, so every one of them must resolve and be reachable from the Director host. When only the seed address is reachable, the target starts cleanly and then waits exactly as in the previous case.

3. The credential was rotated after the target started. Connections are re-authenticated over time, so a key rotated in the provider console stops the flow without a new startup error. Update username and password, then save the target so it reconnects.

4. debug.dont_send_logs is enabled. Records are processed by the pipeline and then discarded before anything is buffered. No counter moves and no error is logged. The only trace is one line at startup, and only when debug.status is enabled as well. Remove the flag when you have finished testing.

5. The broker is not acknowledging writes. With acknowledgments: none and disable_idempotent_write: true, the target asks for no acknowledgment, so a batch the broker discards is still counted as delivered. Use leader or all for anything you need to keep.

6. A TLS setting was accepted but not applied. When both tls.cert_name and tls.key_name are unreadable or sit outside the service root, both are ignored and the connection is made without a client certificate, which a listener that does not insist on one accepts. A misspelled tls.min_tls_version or tls.max_tls_version is read as tls1.2, and a misspelled compression_level falls back to the codec default. Spell these from the tables above, and check the startup logs for invalid tls configuration.

The target never starts and the reason names a setting

These reach you through the same retry loop as a connection failure, so the target keeps retrying a configuration that cannot work until you change it. Nothing is sent while any of them lasts.

Reason text	Fix
address is required for kafka target	Set address. A value of only commas or spaces passes this check and then fails with at least one broker address is required
port must be between 1 and 65535, got 0	Set port within the range, or remove it to use 9092
max_events must be greater than 0, got 0	Set max_events to at least 1, or remove it to use 1000
max_bytes must be between 0 and 104857600, got ...	Lower max_bytes to 104857600 or below
max record batch bytes ... is less than allowed 512	max_bytes is between 1 and 511, which the producer refuses. Use 0 or a value of at least 512
invalid batch_mode: ...	Use individual, json_array, or jsonl, in lower case
acknowledgments must be none, leader or all	Use one of those three words. Numeric forms such as 1 or -1 are not accepted
compression must be none, gzip, snappy, lz4 or zstd	Use one of the listed codecs
algorithm must be plain, scram-sha-256, scram-sha-512 or gssapi	Correct the spelling, and use hyphens rather than underscores
invalid addr: ... or unable to parse port from addr: ...	One entry in address is malformed. Write host, host:port, or a bracketed IPv6 address such as [::1]:9092
kerberos principal is required: ..., kerberos realm is required: ..., kerberos KDC is required: ...	Set kerberos_principal, kerberos_realm and kerberos_kdc, or supply a full krb5.conf through kerberos_config
kerberos keytab is neither an existing file path nor valid base64 keytab content	Point kerberos_keytab at a readable keytab file, or paste the keytab as base64
kerberos login failed for ...	The KDC refused the keytab. Test the same principal and keytab with kinit -kt, confirm TCP 88 is open to the KDC, and check the clock on the Director host
failed to decrypt private key: ...	Set tls.passphrase, or convert the key with openssl pkcs8 -topk8 -v2 aes256
TLS MinVersion must be less than or equal to MaxVersion	tls.min_tls_version is higher than tls.max_tls_version