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

Accept plain JSON for Avro schemas v3.16+

By default, an Avro schema requires every union-typed value to be wrapped in a single-key object that names the union branch. For example, a field declared as ["null", "string"] must be sent as {"string": "hello"} or {"null": null}. This forces HTTP clients to understand Avro’s wire encoding to call your API.

Set payload_encoding: simple_json on the value_schema or the key_schema to let the Kafka Upstream plugin accept plain JSON instead, and resolve union branches against the schema itself:

This resolution applies at every level of the payload, including fields inside nested records, arrays, and maps. Logical types (for example timestamps, decimals, or UUIDs) are passed through unchanged once their union is resolved.

If a record field is omitted from the request body, the plugin falls back to the field’s schema default, if one exists. Otherwise, the field must be nullable, or the plugin rejects the request as missing a required field.

Because an Avro-tagged value like {"string": "hello"} already matches a single branch by name, simple_json accepts it as-is. This lets you migrate clients from avro_json to simple_json one at a time, instead of all at once.

For a sample configuration, see Simple JSON encoding for Avro schemas.

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.

The Kafka Upstream plugin can compress message batches from a producer before sending them to the Kafka broker, using config.compression_type. Compression reduces network bandwidth between Kong Gateway and the broker, broker disk usage, and cross-broker replication cost.

This applies only to the Kong Gateway-to-broker traffic flow. It’s independent of any HTTP-level Content-Encoding between clients and Kong Gateway, and works the same way in both sync and async producer modes.

Note: zstd isn’t available as a producer-side codec, because it requires Kafka Produce API v7+ negotiation. The consume side can already decompress zstd batches written by other producers.

Compression happens at the producer batch level. Kafka Upstream compresses the entire outgoing record batch as a single unit before it’s sent. Compression efficiency improves with batch size, so it’s most effective in async mode, where the plugin already accumulates messages before flushing to the broker. In sync mode, batches are typically smaller, but compression is still available and can be worthwhile for larger payloads.

If config.compression_type is misconfigured or compression fails at runtime, Kafka Upstream logs a warning and sends the batch uncompressed rather than dropping it.

For an example, see Compress messages before sending to Kafka.

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. 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: