Kafka Consume

This plugin consumes messages from Apache Kafka topics and makes them available through HTTP endpoints. For more information, see Kafka topics.

Note: This plugin has the following known limitations:

• Message compression is not supported.
• The message format is not customizable.
• Kong Gateway supports Kafka 4.0 starting from version 3.10.

Kong also provides Kafka plugins for publishing messages:

• Kafka Log
• Kafka Upstream

The plugin supports the following modes of operation:

• http-get: Consume messages via HTTP GET requests (default)
• server-sent-events: Stream messages using server-sent events
• websocket v3.11+: Stream messages over a WebSocket connection

WebSocket mode v3.11+

In websocket mode, the plugin maintains a bi-directional WebSocket connection with the client, allowing for continuous delivery of Kafka messages to the client.

Here’s how it works:

1. Establish a WebSocket connection to the Route where the Kafka Consume plugin is enabled and mode is set to websocket.
2. Kong Gateway continuously streams messages as JSON text frames.
3. Optionally, client sends acknowledgments (client-acks) for each message or batch to enable at-least-once delivery semantics.

This approach provides real-time message flow without the limitations of HTTP polling.

 
sequenceDiagram
    participant Client
    participant Kong as Kong Gateway
    participant Broker as Message Broker

    Client->>Kong: Establish WebSocket connection
    Kong->>Broker: Connect to broker

    loop Continuous message delivery
        Broker->>Kong: Broker message
        Kong->>Client: Stream JSON text frame

        opt If client-acks
            Client->>Kong: Acknowledge message/batch
        end
    end

  

Figure 1: The diagram shows the bi-directional WebSocket flow where the Kafka Consume plugin is running in websocket mode, and messages are streamed as JSON text frames.

This mode provides parity with HTTP-based consumption, including support for:

• Message keys
• Topic filtering
• Kafka authentication and TLS
• Auto or manual offset commits

When running multiple data plane nodes, there is no thread-safe behavior between nodes. In high-load scenarios, you may observe the same message being delivered multiple times across different data plane nodes.

To minimize duplicate message delivery in a multi-node setup, consider:

• Using a single data plane node for consuming messages from specific topics
• Implementing idempotency handling in your consuming application
• Monitoring Consumer Group offsets across your data plane nodes

The Kafka Consume 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 consumer plugin is configured with a schema registry, the following workflow occurs:

 
sequenceDiagram
autonumber
    participant Kafka
    participant Kong as Kafka Consume plugin
    participant Registry as Schema Registry
    participant Client
    
    activate Kafka
    activate Kong
    Kafka->>Kong: Send message
    deactivate Kafka
    Kong->>Kong: Extract schema ID
    activate Registry
    Kong->>Registry: Fetch schema from registry
    Registry-->>Kong: Return schema
    deactivate Registry
    Kong->>Kong: Deserialize using schema
    activate Client
    Kong->>Client: Return response to client
    deactivate Kong
    deactivate Client
  

Configure schema registry

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

For sample configuration values, see:

• Schema registry configuration example

You can use the config.message_by_lua_functions parameter to specify custom Lua code that will filter or transform Kafka messages.

For examples, see the following:

• Transform messages with Lua custom filter code
• Filter messages with Lua custom filter code

The kafka-consume plugin no longer supports scoping to a Service.

Fresh installations

If you try to scope this plugin to a Service on a fresh Kong Gateway instance, a schema violation error will be returned.

Upgrading existing configurations

In traditional mode, Kong Gateway will log an error-level message at startup if a kafka-consume plugin scoped to a Service is detected. The plugin configuration must be updated after startup. Until the configuration is updated, requests to the previous plugin URL will continue to be forwarded to the upstream Service, and responses will be returned to the downstream client as before.

In DB-less mode, if the declarative configuration file contains a kafka-consume plugin scoped to a Service, Kong Gateway will fail to start. In this case, proceed as follows:

1. Update the declarative configuration 1. Restart Kong Gateway.

In hybrid mode, if a kafka-consume plugin scoped to a Service exists in PostgreSQL, upgrading and restarting the control plane will succeed. Then, to avoid validation errors on sync, follow these steps:

1. Update the plugin configuration.
2. Update and restart the data plane.

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