Kafka Upstream

This plugin converts requests into Apache Kafka messages and publishes them to a specified Kafka topic.
For more details, see Kafka topics.

Kong Gateway also offers a separate Kafka Log plugin for streaming logs to Kafka topics.

This plugin uses the lua-resty-kafka client.

When encoding request bodies, several things happen:

• For requests with a content-type header of application/x-www-form-urlencoded, multipart/form-data, or application/json, this plugin passes the raw request body in the body attribute, and tries to return a parsed version of those arguments in body_args. If this parsing fails, an error message is returned and the message is not sent.
• If the content-type is not text/plain, text/html, application/xml, text/xml, or application/soap+xml, then the body will be base64-encoded to ensure that the message can be sent as JSON. In such a case, the message has an extra attribute called body_base64 set to true.

The Kafka Upstream plugin supports integration with Confluent Schema Registry for AVRO and JSON schemas.

Schema registries provide a centralized repository for managing and validating schemas for data formats like AVRO and JSON. Integrating with a schema registry allows the plugin to validate and serialize/deserialize messages in a standardized format.

Using a schema registry with Kong Gateway provides several benefits:

• Data validation: Ensures messages conform to a predefined schema before being processed.
• Schema evolution: Manages schema changes and versioning.
• Interoperability: Enables seamless communication between different services using standardized data formats.
• Reduced overhead: Minimizes the need for custom validation logic in your applications.

To learn more about Kong’s supported schema registry, see:

• Confluent Schema Registry Documentation
• AVRO Specification

How schema registry validation works

When a producer plugin is configured with a schema registry, the following workflow occurs:

 
sequenceDiagram
autonumber
    participant Client
    participant Kong as Kafka Upstream plugin
    participant Registry as Schema Registry
    participant Kafka
    
    activate Client
    activate Kong
    Client->>Kong: Send request
    deactivate Client
    activate Registry
    Kong->>Registry: Fetch schema from registry
    Registry-->>Kong: Return schema
    deactivate Registry
    Kong->>Kong: Validate message against schema
    Kong->>Kong: Serialize using schema
    activate Kafka
    Kong->>Kafka: Forward to Kafka
    deactivate Kong
    deactivate Kafka
  

If validation fails, the request is rejected with an error message.

Configure schema registry

To configure Schema Registry with the Kafka Upstream plugin, use the config.schema_registry parameter in your plugin configuration.

For sample configuration values, see:

• Schema registry configuration example

The Kafka Upstream plugin can forward HTTP request headers as Kafka record headers, which are per-record key/value metadata that lives alongside the message key and value. This lets consumers read routing, tracing, or tenancy context without parsing the message payload.

Configure the config.headers block to control which headers are forwarded:

Use config.headers.name_mappings to rename an HTTP header to a different Kafka record header key.

Use config.headers.repeated_headers_behavior to control how duplicate HTTP headers are handled: retain_duplicates (default) creates a separate record header per value, take_first uses only the first value, and concatenate_by_comma joins all values with a comma.

Note: The config.forward_headers setting embeds request headers inside the message body. config.headers is a separate configuration block that sets native Kafka record headers on the produced record.

By default, when producing a message fails, the plugin returns a generic error to the HTTP client and logs the real cause in the Kong Gateway logs:

{"message": "Bad Gateway", "error": "could not send message to topic"}

To include the detailed Kafka client error in the response instead, set config.error_handling.return_error_message to true (false by default). This lets you see the exact broker rejection (for example, TopicAuthorizationFailed) without checking the Kong Gateway logs.

Warning: Do not enable this in production. The detailed error may expose internal details like topic names and ACL denials.

Known limitations:

1. Message compression is not supported.
2. In Kong Gateway 3.9 or earlier, the message format is not customizable.

The Kafka Upstream plugin supports the following SASL authentication mechanisms for broker connections via authentication.mechanism: