Kafka publisher

Publishes every incoming record as a message to a Kafka topic. Accepts either a ready-made EventData (key, headers, timestamp, binary payload) or a TextRecord whose text becomes the payload.

Pins

Pin	Kind	Types	Description
Input	input	Viaducta.Plugins.Event.Types.EventData, Viaducta.Plugins.TextPlugin.Types.TextRecord	Record to publish.
Continue	continue	same as input	Called with the unchanged input record after the message was handed to the Kafka client.

Configuration

Field	Type	Default	Description
Topic	string (max 255)		Required. Topic to publish to. Created at flow start if it does not exist.
BootstrapServers	string (max 512)		Required. Broker address as host:port; several brokers comma separated.
Username	string (max 255)		SASL user name. See Notes: currently has no effect.
Password	password (max 255)		SASL password. See Notes: currently has no effect.
DefaultReplicationFactor	integer (min 1)	1	Replication factor used only when the plugin creates the topic.
GroupId	string (max 255)		Not used yet: the consumer always uses a random group id (see Notes).
SecurityProtocol	dropdown: 0 Plaintext, 1 Ssl, 2 SaslPlaintext, 3 SaslSsl	0	Not used yet: the client always connects in plaintext (see Notes).
SaslMechanism	dropdown: 1 Plain, 2 ScramSha256, 3 ScramSha512	1	Visible with SecurityProtocol 2 / 3. Not used yet.

Behaviour

• At flow start the producer is created and the topic is checked (metadata request with a 1 second timeout) and created with DefaultReplicationFactor if missing; a creation error is only logged. The plugin keeps the flow running (KeepAlive).
• Message built from the input:
  • EventData: Id → message key, Data → value, Headers → message headers (values encoded as UTF-8, null becomes empty), Time → message timestamp (current UTC time if null).
  • TextRecord: Value → value (UTF-8 bytes; null value gives an empty message), key = new random GUID (32 hex characters), timestamp = now, no headers. Name and ContentType are not sent.
  • Any other type raises an error.
• The message is queued in the Kafka client (fire-and-forget) and Continue is called immediately with the input record.

Notes

• No delivery confirmation: Continue is called before the broker confirmed the message and delivery errors are not reported to the flow. Messages still queued when the flow stops can be lost.
• Authentication: Username / Password are set as SASL credentials, but the security protocol and SASL mechanism are not set, so the client connects in plaintext without authentication. TLS is not supported.
• The topic is fixed per component; the record cannot choose the topic.
• To control the key or add headers, build an EventData (for example in a script) instead of sending a TextRecord.

Example

{ "Topic": "orders", "BootstrapServers": "kafka1:9092", "DefaultReplicationFactor": 1 }

Http listener → Json serializer (TextRecord) → Kafka publisher → Http response