Kafka

Writes each record to a Kafka topic.

Smeltr manages the producer, including its protection against duplicates caused by producer retries.

Example

Writes to a plaintext test broker. Replace its address with one reachable from the worker, and keep input and output topics separate unless you mean to build a feedback path.

{
  "id": "events-out",
  "type": "kafka",
  "brokers": "localhost:9092",
  "defaultTopic": "smeltr-events-out"
}

Before you start

  • Create the output topic.
  • Allow workers to reach every broker address Kafka advertises.
  • Give the Kafka account permission to write to the topic.
  • For an authenticated broker, configure Kafka security. Select a connection identity for a private CA or a client certificate.

Settings

It also takes the settings every destination takes.

brokers string required

The brokers a worker first connects to, host:port, separated by commas.

Examples
kafka-1:9092,kafka-2:9092broker.example.com:9093

defaultTopic string required

The topic for a record with no __destination.topic.

Examples
orders-cleansecurity.firewall.v1

format string default json

What each message holds.

json
Writes the record as JSON.
text
Writes one field's text, from field.
bytes
Writes each source record's bytes as read, before pipeline processing, once per record: __source.bytes unless field names another bytes field.
avro
Writes Avro messages, registering the schema in the schema registry.

schemaRegistry object default none

The schema registry the Avro schema is registered in. Set url. If the registry needs a sign-in, also set username and password. Needed for format avro.

Examples
{"url": "https://registry.example.com"}{"url": "https://registry.example.com","username": "smeltr","password": {"secret": "registry","version": 1}}

field string default _raw for text, __source.bytes for bytes

The field each text or bytes record writes. Only with format text or bytes.

Example
message

identity identity default none

The connection identity it connects under, for a private CA or a client certificate. Without one, no certificate is offered and the public roots, Mozilla's and the host's, are trusted.

Examples
kafka-clientkafka-mtls

Advanced: Delivery and client properties

compression string default none

How it compresses each batch of messages.

none
Sends it uncompressed.
gzip
Compresses it with gzip.
snappy
Compresses it with Snappy.
lz4
Compresses it with LZ4.
zstd
Compresses it with Zstandard.

stallAfterSecs integer default 30

Seconds one batch may take to deliver before it counts as stalled. 1 to 120.

Examples
60120

bufferBytes size default 64 MiB

How much undelivered output it may hold. The Kafka client holds as much again. 4 MiB to 1 GiB.

Examples
256 MiB16 MiB1 GiB

properties map default none

Other Kafka client properties, each name to its value. See Kafka security. A property Smeltr sets itself, such as bootstrap.servers, is refused, and a secret one belongs in credentials.

Examples
security.protocol = SASL_SSLsecurity.protocol = SASL_SSLsasl.mechanism = SCRAM-SHA-512sasl.username = smeltr

credentials map default none

Kafka client properties whose values are secret, each name to the credential holding its value.

Examples
sasl.password = KAFKA_PASSWORDsasl.oauthbearer.client.secret = KAFKA_CLIENT_SECRET

Fields it reads

FieldTypeHolds
__destination.topicstrThe Kafka topic or Event Hub a destination writes the record to. Null means the default value for that destination.

Verify

Send a record you can recognize through the source, consume the output topic, and check the record's fields, not just that the topic has messages. The tutorial's delivery check shows the commands.

Troubleshooting

Timeout or authorization error
Check advertised addresses, that the topic exists, write permissions and Kafka security.

See also