Kafka Trigger

Consume messages from a Kafka topic

Kafka Trigger

AI-generated

Overview

This n8n node, Kafka Trigger, is designed to consume messages from an AWS MSK (Managed Streaming for Apache Kafka) topic. It supports advanced features such as Confluent Schema Registry integration, flexible consumer group management, and various consumption options. The node is typically used to trigger workflows in response to new messages arriving on a Kafka topic, making it ideal for event-driven architectures, real-time data processing, and integrating Kafka streams with other systems via n8n.

Practical examples:

  • Start a workflow whenever a new order message arrives in a Kafka topic.
  • Process log or telemetry data streamed into Kafka in near real-time.
  • Integrate microservices by consuming events published to Kafka topics.

Properties

Name Type Meaning
Topic String Name of the Kafka topic to consume messages from.
Group ID String Identifier for the Kafka consumer group.
Use Schema Registry Boolean Whether to use Confluent Schema Registry for message decoding.
Schema Registry URL String URL of the schema registry (required if "Use Schema Registry" is enabled).
Options Collection Additional configuration options (see below).
└ Allow Topic Creation Boolean Allow sending messages to a previously non-existing topic.
└ Auto Commit Threshold Number Commit offsets after resolving this number of messages.
└ Auto Commit Interval Number Commit offsets after this period (milliseconds).
└ Heartbeat Interval Number Interval for heartbeats to keep the consumer session active (milliseconds).
└ Max Number of Requests Number Maximum concurrent requests allowed; unlimited if not set.
└ Read Messages From Beginning Boolean If true, start reading from the beginning of the topic.
└ JSON Parse Message Boolean Attempt to parse the message value as JSON.
└ Only Message Boolean If enabled (and JSON Parse Message is true), return only the parsed message property.
└ Return Headers Boolean Include Kafka message headers in the output.
└ Session Timeout Number Time to await a response before timing out (milliseconds).

Output

The node emits items with the following structure in the json field:

  • If Only Message is enabled (and JSON Parse Message is true):

    { /* parsed message object */ }

  • Otherwise:

    {
  "message": /* string, object, or decoded value */,
  "topic": "topic-name",
  "headers": { /* optional, present if Return Headers is enabled */ }
}

  • message: The content of the Kafka message. This may be a string, a parsed object (if JSON Parse Message is enabled), or a decoded value (if Schema Registry is used).

  • topic: The name of the topic from which the message was consumed.

  • headers: (Optional) An object containing Kafka message headers, included if "Return Headers" is enabled.

Note: The node does not output binary data.

Dependencies

  • External Services:
    • AWS MSK (Kafka) cluster.
    • (Optional) Confluent Schema Registry, if using Avro/Protobuf encoded messages.
  • Credentials:
    • n8n credential type: kafkaawscred (must include brokers, clientId, ssl, region).
  • Node.js Packages:
    • kafkajs
    • @kafkajs/confluent-schema-registry
    • @jm18457/kafkajs-msk-iam-authentication-mechanism
  • n8n Configuration:
    • Ensure the node has access to the required credentials and network connectivity to Kafka brokers and (optionally) the schema registry.

Troubleshooting

Common Issues:

  • Connection Errors:
    • "Failed to connect to broker" — Check your broker addresses, network/firewall settings, and credentials.
  • Authentication Failures:
    • "SASL authentication failed" — Verify IAM permissions and that the correct region is specified.
  • Schema Registry Errors:
    • "Could not decode message" — Ensure the Schema Registry URL is correct and accessible, and that the message format matches the expected schema.
  • Message Parsing Errors:
    • "Unexpected token..." — If "JSON Parse Message" is enabled, ensure incoming messages are valid JSON.
  • Timeouts:
    • "Session timeout" or "Heartbeat missed" — Adjust the "Session Timeout" and "Heartbeat Interval" options according to your Kafka cluster's configuration.

How to resolve:

  • Double-check all input properties, especially connection details and credentials.
  • If using Schema Registry, confirm its URL and accessibility.
  • For parsing issues, disable "JSON Parse Message" to inspect raw message content.

Links and References

Discussion