Kafka Producer

Sends messages to a Kafka topic

Kafka Producer

AI-generated

Overview

This node, Kafka Producer, allows you to send messages to an AWS MSK (Managed Streaming for Apache Kafka) topic from n8n workflows. It supports sending either the incoming JSON data or a custom message, with optional support for Confluent Schema Registry and message keys. You can also add custom headers and configure advanced producer options such as acknowledgements, compression, and timeouts.

Common scenarios:

  • Integrating n8n with real-time data pipelines by publishing workflow results to Kafka topics.
  • Sending event notifications, logs, or processed data to downstream consumers via Kafka.
  • Publishing Avro-encoded messages using a schema registry for compatibility with data lakes or analytics platforms.

Practical examples:

  • Forwarding webhook payloads to a Kafka topic for further processing.
  • Sending alerts or monitoring data to a Kafka stream.
  • Publishing user activity events to a Kafka topic for analytics.

Properties

Name Type Meaning
Topic string Name of the Kafka topic to publish to.
Send Input Data boolean If true, sends the input JSON data received by the node; otherwise, uses the Message field.
Message string The message to be sent if "Send Input Data" is false.
JSON Parameters boolean If true, headers are provided as a JSON object; otherwise, use the UI collection.
Use Schema Registry boolean Whether to encode the message using a Confluent Schema Registry.
Schema Registry URL string URL of the schema registry (required if "Use Schema Registry" is true).
Use Key boolean Whether to include a message key in the Kafka message.
Key string The message key (required if "Use Key" is true).
Event Name string Namespace and name of the schema in the registry (required if "Use Schema Registry" is true).
Headers fixedCollection Custom headers as key-value pairs (used if "JSON Parameters" is false).
Headers (JSON) json Header parameters as a flat JSON object (used if "JSON Parameters" is true).
Options collection Advanced producer options: Acks, Compression, Timeout.

Output

The node outputs an array of result objects from the Kafka producer's sendBatch operation. Each output item contains information about the result of the batch send operation. If no response is returned, it outputs { success: true }. In case of error and "Continue On Fail" is enabled, it outputs { error: <error message> }.

Example output:

[
  {
    "topicName": "example-topic",
    "partition": 0,
    "errorCode": 0,
    "baseOffset": "42",
    "logAppendTime": "-1",
    "logStartOffset": "0"
  }
]

or, on success with no details:

[
  { "success": true }
]

or, on error (if "Continue On Fail" is enabled):

[
  { "error": "Error message here" }
]

If the message is encoded using the schema registry, the actual message value sent to Kafka will be binary, but the node output itself remains JSON as above.

Dependencies

  • External Services:

    • AWS MSK (Kafka cluster)
    • (Optional) Confluent Schema Registry

  • Credentials Required:

    • kafkaawscred (must provide brokers, clientId, ssl, region)

  • Node.js Packages:

    • kafkajs
    • @kafkajs/confluent-schema-registry
    • @jm18457/kafkajs-msk-iam-authentication-mechanism
    • n8n-workflow (for error handling)

  • n8n Configuration:

    • Ensure the required credentials are set up in n8n under the name kafkaawscred.

Troubleshooting

Common Issues:

  • Invalid Schema Registry configuration:
    Error: "Verify your Schema Registry configuration"

    • Check that the Schema Registry URL and Event Name are correct and accessible.
    • Ensure the message is valid JSON and matches the schema.

  • Headers must be a valid json:
    Error: "Headers must be a valid json"

    • When using "Headers (JSON)", ensure the input is a valid flat JSON object.

  • Connection errors:

    • Ensure the Kafka brokers, client ID, SSL, and region are correctly configured in credentials.
    • Network/firewall issues may prevent connecting to the Kafka cluster.

  • Missing required fields:

    • If "Use Key" or "Use Schema Registry" is enabled, make sure to fill in the required "Key", "Schema Registry URL", and "Event Name" fields.

  • No output or empty response:

    • If the Kafka cluster does not return any response, the node outputs { success: true }.

Links and References

Discussion