Kafka subscriber

Subscribes to one Kafka topic and emits every received message as an EventData record. Use it as the starting point of a flow that reacts to Kafka events.

Pins

Pin	Kind	Types	Description
Start	start		Starts the consumer. Called by the system at flow start when unconnected. Further calls are ignored once the consumer runs.
Continue	continue	Viaducta.Plugins.Event.Types.EventData	Called once for every message received from the topic.

Configuration

Field	Type	Default	Description
Topic	string (max 255)		Required. Name of the topic to consume. Created on start if it does not exist.
BootstrapServers	string (max 512)		Required. Broker address as host:port; several brokers comma separated (kafka1:9092,kafka2:9092).
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 has to create 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

• The plugin keeps the flow running (KeepAlive) until the flow is stopped.
• On Start it connects to the brokers, checks whether Topic exists (metadata request with a 1 second timeout) and creates it with DefaultReplicationFactor if missing. A topic creation error is only logged. An empty Topic stops the flow with an unrecoverable error.
• It then subscribes in the background and calls Continue for every message, one after another (the next message is read after the downstream processing of the previous one returns).
• Every message becomes an EventData:
  • Id - message key (string, may be null),
  • Data - message value as raw bytes (byte[]),
  • Time - message timestamp in UTC,
  • Headers - message headers as a string dictionary; header values are decoded as UTF-8.
• Every message gets new metadata; its CorrelationId is the message key.
• An error while reading from Kafka is logged and reading resumes after one second. An error in the downstream processing of a message is logged and the message is skipped; the subscription keeps running.

Notes

• Consumer group: every flow start uses a new random consumer group id. Consumers are therefore never shared/load-balanced between flow instances (each instance receives all messages), committed offsets are never reused, and the consumer starts at the end of the topic: only messages produced after the subscription are received. Messages produced while the flow was stopped are not delivered.
• Authentication: Username / Password are passed as SASL credentials, but the security protocol and SASL mechanism are not set, so the client connects in plaintext without authentication. Only brokers that accept unauthenticated plaintext connections work. TLS is not supported.
• Only one topic per component. Partition count of an auto-created topic is not configurable.
• Data is binary. To work with text or JSON, convert the bytes downstream (for example with a script or a converter component).

Example

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

Kafka subscriber (Continue) → Script (decode Data as UTF-8) → Json deserializer → ...