Kafka Producer

Connection

The Kafka connection to use.

Evaluated for each input row

Mode

The producer can operate in two modes:

• In synchronous mode, the step waits for the result of each written message before processing the next row.
  This mode makes the meta step result available and supports error handling in the usual way.
• In asynchronous mode, the step sends out messages as quickly as possible without waiting for confirmation from the broker.
  This is typically very fast, but there is no meta result in asynchronous mode, and per-row error handling is not supported. Delivery errors are logged, but no error rows are generated. This mode has ‘fire-and-forget’ semantics unless you are using it with transactions enabled.

Evaluated for each input row

Topic

The topic to write messages to.

Evaluated for each input row

Key

The key of the current message. Leave as nil to not use a key.

Both string and binary data types are supported.

Evaluated for each input row

Message

The message to write.

Both string and binary data types are supported.

Evaluated for each input row

Timestamp

The timestamp of the current message. Leave as nil to let the broker assign a timestamp.

You can use the following datatypes:

• datetime
• long - interpreted as unix timestamp in ms

Evaluated for each input row

partition

The partition to write the current message to. Use nil to let the broker assign a partition.

Evaluated for each input row

headers

Any headers to set for the message.

Each header entry in the list is a dict with :name and :value keys.

Header names are strings, header values can be of type string or binary.

Evaluated for each input row

Client ID

The Client ID of the producer.

Evaluated for each input row

Use Transactions

When true, the producer uses transactions to write messages to the broker.

Make sure to review Kafka transactional semantics and configuration before using this feature.

Be sure to set a transactional.id in properties.

The producer wraps messages in transactions. Transactions are committed when any of the following conditions are met:

• Last row has been processed
• Commit size is reached
• Connection, ClientID, or Properties settings have changed

Note that any change to the properties setting causes a commit of an open transaction.

The transactional.id property can be conveniently used to achieve samantic commits. If you have several consecutive messages - potentially for different topics - regarding a single customer_id for example, construct a transactional.id that incorporates the customer_id and is therefore uniqe to your group of rows. Any consecutive group of messages with the same transactional.id will be committed as a separate transaction.

Supported Modes

Transactions are designed to be used in asynchronous mode as this mode gives more useful delivery semantics:

• Write errors in asynchronous mode are discovered when the transaction is committed. In this case a single error row is generated and it stands for the entire transaction.

• Write errors in synchronous mode abort the current transaction and generate an error row immediately. The next incoming row automatically opens a new transaction. These semantics are useful only if errors remain unhandled and cause the flow to abort.

Evaluated once when step initializes

Commit Size

Only used when transactions are enabled.

The following values are possible:

• When set to nil or 0, the step does not commit transactions based on message count
• When set to a value greater than zero, the step commits the current transaction after it reaches the given amount of messages

Any other values produce an error.

Evaluated once when step initializes

Properties

Additional producer properties to use.

See Kafka Configuration documentation for details.

Any properties given override properties already defined in the Kafka connection.

If you supply a dict, the keys and values are interpreded as property names and values.

If you supply a list, each item must be of the form: {:name string, :value string}.

meta

In synchronous mode the step provides message meta-data for each row as provided by the kafka broker.

The meta-data has the following form:

{
  :topic 'topic name',
  :has_timestamp true,
  :offset 4321,
  :timestamp 1610895234545,
  :has_offset true,
  :serialized_key_size -1,
  :partition 0,
  :serialized_value_size 224
}