Apache Kafka

Apache Kafka is a distributed event-streaming platform used to publish, retain, and process records across applications and services.

The Apache Kafka Node publishes records, performs bounded consumer-group reads, and inspects topic metadata. Use it when a Workflow needs to exchange JSON, text, or binary data with a Kafka cluster.

Revision History

0.0.0.3 - Corrected the display of Enums.

0.0.0.2 - Corrected the response shape to use a single Response envelope.

0.0.0.1 - Initial release.

Connection

Create an Apache Kafka Connection to define the broker endpoints, default topic and consumer group, and the security settings used by every method.

Property	Type	Description
Bootstrap Servers	String	Required comma-separated host:port endpoints used for initial broker discovery.
Client ID	String	Required client identifier included in Kafka broker requests and diagnostics. The default is flowgear-apache-kafka.
Default Topic	String	Optional exact topic name used when a method omits Topic.
Default Consumer Group ID	String	Optional consumer group used when Consume Messages omits Group ID.
Security Protocol	Enum	Select Plaintext, Ssl, SaslPlaintext, or SaslSsl.
SASL Mechanism	Enum	Select Plain, ScramSha256, or ScramSha512 when using a SASL security protocol.
SASL Username	String	Required principal name when Security Protocol is SaslPlaintext or SaslSsl.
SASL Password	Masked	Required secret when Security Protocol is SaslPlaintext or SaslSsl.
CA Certificate PEM	String	Optional PEM-encoded certificate authority used by TLS transports. The platform trust store is used when this is blank.
Client Certificate PEM	String	Optional PEM-encoded client certificate for mutual TLS. Supply this together with Client Private Key PEM.
Client Private Key PEM	Masked	PEM-encoded private key paired with Client Certificate PEM for mutual TLS.
Client Key Password	Masked	Optional passphrase for an encrypted Client Private Key PEM.

Setup Notes

• Ensure the Runtime or DropPoint can resolve and reach at least one endpoint in Bootstrap Servers.
• Configure broker ACLs for the operations the Workflow performs. Producing requires topic write access, consuming requires topic and consumer-group read access, and metadata methods require permission to inspect the relevant topics.
• The Node does not automatically create topics. Create each topic and configure its partitions and replication before running the Workflow.
• Use Ssl for TLS without SASL authentication. Use SaslSsl when the cluster requires SASL authentication over TLS.
• SaslPlaintext authenticates but does not encrypt network traffic. Use it only on a trusted private network.
• SaslPlaintext and SaslSsl require SASL Username and SASL Password.
• TLS Connections verify the certificate chain and broker hostname. Supply CA Certificate PEM when the broker certificate is not trusted by the platform trust store.
• For mutual TLS, supply both Client Certificate PEM and Client Private Key PEM. Supply Client Key Password only when the private key is encrypted.
• Use the Connection test to verify broker discovery and authentication before adding the Node to a Workflow.

Methods

This Node exposes methods for publishing records, performing bounded consumer-group reads, and inspecting topics and partitions.

Produce Messages

Use this method to publish one Kafka record for each row in Items. The first record is sent while the Node invocation starts, so initial validation, connection, and delivery errors surface before output streaming begins.

Parameter	Type	Description
Connection	Connection	Apache Kafka Connection details.
Topic	String	Exact destination topic. If this is blank, the Node uses Default Topic from the Connection.
Value Format	Enum	Select Json, Utf8Text, or Base64 for record values. The default is Json.
Key Format	Enum	Select Utf8Text or Base64 for record keys. The default is Utf8Text.
Header Value Format	Enum	Select Utf8Text or Base64 for header values. The default is Utf8Text.
Items	Array	Required stream of records to publish. See the Produce Item Properties table below.
Delivery Timeout Seconds	Integer	Maximum time to wait for each broker delivery acknowledgement. Enter a value from 1 to 300. The default is 30.

Return	Type	Description
Response	Array	One delivery row per input row. See the Produce Response Properties table below.

Produce Item Properties

Property	Type	Description
Key	String	Optional record key encoded using Key Format.
Value	Object	Record value encoded using Value Format. This is required unless IsTombstone is true.
IsTombstone	Boolean	Set to true to publish a Kafka null value. When enabled, Value must be omitted or JSON null.
Partition	Integer	Optional zero-based destination partition. Leave this blank to let Kafka select the partition.
Headers	Array	Optional ordered header rows. Duplicate header keys and null header values are preserved.

Each Headers row contains a required string Key and an optional Value encoded using Header Value Format.

Produce Response Properties

Property	Type	Description
Topic	String	Topic that accepted the record.
Partition	Integer	Partition that accepted the record.
Offset	Integer	Kafka offset assigned to the record.
TimestampUtc	DateTime	UTC record timestamp returned by Kafka, when available.
Flowgear	Object	Contains IsSuccess, Message, and a copy of the input row in Request.
KafkaErrorCode	String	Kafka error code for a recoverable per-item provider failure, when available.
IsFatal	Boolean	Indicates whether Kafka classified the provider failure as fatal, when available.
IsRetriable	Boolean	Indicates whether Kafka classified the provider failure as retriable, when available.

After startup, a recoverable per-item error returns a failed Flowgear result and processing continues with the next row. Fatal producer errors stop the invocation.

Consume Messages

Use this method to read a bounded sequence of records through one Kafka consumer group. The Node confirms the topic and performs the first poll while the invocation starts, so initial provider errors surface before output streaming begins.

Parameter	Type	Description
Connection	Connection	Apache Kafka Connection details.
Topic	String	Exact source topic. If this is blank, the Node uses Default Topic from the Connection.
Group ID	String	Consumer group ID. If this is blank, the Node uses Default Consumer Group ID from the Connection.
Value Format	Enum	Select Json, Utf8Text, or Base64 for record values. The default is Json.
Key Format	Enum	Select Utf8Text or Base64 for record keys. The default is Utf8Text.
Header Value Format	Enum	Select Utf8Text or Base64 for header values. The default is Utf8Text.
Auto Offset Reset	Enum	Select Earliest, Latest, or Error for a group without a valid committed offset. The default is Earliest.
Isolation Level	Enum	Select ReadCommitted or ReadUncommitted to control visibility of transactional records. The default is ReadCommitted.
Offset Commit Mode	Enum	Select CommitAfterDelivery or DoNotCommit. The default is CommitAfterDelivery.
Maximum Messages	Integer	Maximum records returned by this invocation. Enter a value from 1 to 10000. The default is 100.
Idle Timeout Seconds	Integer	Ends the invocation successfully when no record arrives during this interval. Enter a value from 1 to 300. The default is 5.
Metadata Timeout Seconds	Integer	Maximum time to wait for initial topic discovery. Enter a value from 1 to 300. The default is 10.

Return	Type	Description
Response	Array	One row per consumed Kafka record. An idle topic returns no rows. See the Consumed Record Properties table below.

Consumed Record Properties

Property	Type	Description
Topic	String	Topic containing the record.
Partition	Integer	Partition containing the record.
Offset	Integer	Record offset within the partition.
TimestampUtc	DateTime	UTC record timestamp, when available.
TimestampType	String	Kafka timestamp classification, such as CreateTime.
Key	String	Record key decoded using Key Format, or null when the record has no key.
Value	Object	Record value decoded using Value Format, or null for a tombstone.
IsTombstone	Boolean	Indicates whether the Kafka record has a null value.
Headers	Array	Ordered header rows containing Key and Value. Duplicate keys and null values are preserved.

CommitAfterDelivery commits a record when downstream processing requests the next row or completes the stream. If processing abandons or fails on the current row, Kafka can deliver that record again. DoNotCommit leaves the consumer group offsets unchanged.

List Topics

Use this method to list topics visible to the Connection. Kafka provider errors surface to the Workflow; a cluster with no visible topics returns no rows.

Parameter	Type	Description
Connection	Connection	Apache Kafka Connection details.
Include Internal Topics	Boolean	Includes topics whose names begin with __. The default is false.
Metadata Timeout Seconds	Integer	Maximum time to wait for cluster metadata. Enter a value from 1 to 300. The default is 10.

Return	Type	Description
Response	Array	One provider-shaped row per visible topic.

Topic Properties

Property	Type	Description
Topic	String	Exact topic name.
PartitionCount	Integer	Number of partitions in the topic.

Describe Topic

Use this method to inspect partition leadership and replica placement for one topic. A missing topic or Kafka provider error surfaces to the Workflow.

Parameter	Type	Description
Connection	Connection	Apache Kafka Connection details.
Topic	String	Exact topic to inspect. If this is blank, the Node uses Default Topic from the Connection.
Metadata Timeout Seconds	Integer	Maximum time to wait for topic metadata. Enter a value from 1 to 300. The default is 10.

Return	Type	Description
Response	Array	One provider-shaped row per topic partition.

Partition Properties

Property	Type	Description
Topic	String	Exact topic name.
Partition	Integer	Zero-based partition number.
LeaderBrokerId	Integer	Broker ID of the current partition leader.
ReplicaBrokerIds	Array	Broker IDs hosting replicas of the partition.
InSyncReplicaBrokerIds	Array	Broker IDs for replicas currently in sync with the leader.

Usage Notes

• Produce Messages and Consume Messages are bounded Workflow operations. This Node does not run as a perpetual Kafka event trigger.
• Kafka preserves ordering within a partition, not across all partitions in a topic.
• Producer idempotence reduces duplicate writes within one producer session. A Workflow retry can still publish a duplicate business event.
• Json requires valid JSON data, Utf8Text uses strict UTF-8, and Base64 preserves arbitrary bytes. A record that does not match the selected format causes an error instead of being silently changed.
• Tombstone records carry a Kafka null value and are commonly used by compacted topics to represent deletion.
• The Node does not provide Schema Registry serializers, Kafka transactions, or consume-transform-produce exactly-once processing.