Kafka Decimal Trigger icon

Kafka Decimal Trigger

Consume messages from a Kafka topic with decimal support

Kafka Decimal Trigger

AI-generated

Overview

The Kafka Decimal Trigger node for n8n allows you to consume messages from a Kafka topic, with special support for decimal data types (including Avro logical types via Confluent Schema Registry). This node is useful in workflows where you need to react to new messages on a Kafka topic—such as ingesting event streams, processing logs, or integrating real-time data pipelines. It is especially beneficial when working with financial, scientific, or other data that requires precise decimal handling.

Example scenarios:

  • Triggering workflows when new orders are placed in an e-commerce system (messages sent to a Kafka topic).
  • Processing IoT sensor data streams with decimal values.
  • Integrating with systems using Avro schemas and Confluent Schema Registry for message serialization.

Properties

Name Type Meaning
Topic String Name of the Kafka topic to consume messages from.
Group ID String Consumer group ID for Kafka; used to manage message consumption state.
Use Schema Registry Boolean Whether to use Confluent Schema Registry for decoding Avro/serialized messages.
Schema Registry URL String URL of the schema registry (required if "Use Schema Registry" is enabled).
Options → Allow Topic Creation Boolean Allow sending messages to a previously non-existing topic.
Options → Auto Commit Threshold Number Commit offsets after resolving this number of messages.
Options → Auto Commit Interval Number Commit offsets after this period (in milliseconds).
Options → Heartbeat Interval Number Interval for consumer heartbeats (in milliseconds); must be less than session timeout.
Options → Max Number of Requests Number Maximum number of concurrent requests; 0 means no limit.
Options → Read Messages From Beginning Boolean If true, read messages from the beginning of the topic.
Options → JSON Parse Message Boolean Attempt to parse the message value as JSON.
Options → Only Message Boolean If enabled (and JSON Parse Message is true), return only the message property.
Options → Return Headers Boolean Include Kafka message headers in the output.
Options → Session Timeout Number Time to await a response before timing out (in milliseconds).

Output

The node emits items with the following structure:

  • If Only Message is enabled (and JSON Parse Message is true):
    The output is the parsed message content directly.

  • Otherwise, the output is an object with:

    • message: The message value (string, object, or decoded Avro, depending on options).
    • topic: The topic name.
    • headers (optional): An object containing Kafka message headers (if "Return Headers" is enabled).

Example output (with headers and full object):

{
  "message": {
    "orderId": 123,
    "amount": 45.67
  },
  "topic": "orders",
  "headers": {
    "correlation-id": "abc123"
  }
}

Example output (only message):

{
  "orderId": 123,
  "amount": 45.67
}

If binary data is present in the message, it will be handled according to the parsing and schema registry settings, but the node does not emit n8n binary fields.

Dependencies

  • External Services:

    • Kafka cluster (brokers, topics, etc.).
    • (Optional) Confluent Schema Registry (if "Use Schema Registry" is enabled).

  • Credentials:

    • n8n Kafka credentials (including brokers, clientId, SSL, authentication, username/password, SASL mechanism as needed).

  • Node.js Packages:

    • kafkajs
    • @kafkajs/confluent-schema-registry
    • Custom decimal logical type handler (./logical-types/decimal).

  • n8n Configuration:

    • Ensure the node has access to required environment variables and network connectivity to Kafka and Schema Registry endpoints.

Troubleshooting

Common Issues:

  • Authentication errors:
    If username/password or SASL mechanism are missing or incorrect, the node will throw an error indicating missing credentials.
  • Schema Registry decoding failures:
    If the message cannot be decoded using the provided schema registry URL, an error will be logged, and the message may not be processed as expected.
  • Connection issues:
    Errors connecting to Kafka brokers or Schema Registry (e.g., network/firewall issues, wrong URLs).
  • JSON parsing errors:
    If "JSON Parse Message" is enabled and the message is not valid JSON, parsing will fail silently (the original string will be returned).

Error Messages:

  • "Username and password are required for authentication":
    Ensure all authentication fields are filled in the credentials.
  • [KafkaDecimalTrigger] Message failed to decode:
    Check your schema registry URL and ensure the message format matches the expected schema.

Links and References

Discussion