Kafka Consume Policy

Overview Configuration reference

Implementation details

The Policy supports the following modes of operation, set with config.mode:

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

WebSocket mode

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

Here’s how it works:

The client establishes a WebSocket connection to the endpoint where the Kafka Consume Policy is enabled with the mode set to websocket.
AI Gateway continuously streams messages as JSON text frames.
Optionally, the 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 AI 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 Policy 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

The Policy serves messages on the Route of the entity it’s attached to. Reference it from the policies array on an AI Model, AI Agent, or AI MCP Server that defines config.route, then send requests to that path. A global Kafka Consume Policy runs on every AI Gateway Route, so requests only reach it where a Route already exists.

In http-get mode, send a GET request to the path with no query parameters. The Policy returns the records available for every topic in config.topics, keyed by topic name and then by partition:

{
  "ai-events": {
    "partitions": {
      "0": {
        "high_watermark": 1,
        "last_stable_offset": 1,
        "errcode": 0,
        "records": [
          {
            "value": {"event": "prompt_received", "model": "gpt-4o"},
            "key": "",
            "timestamp": 1787305085262,
            "offset": 0
          }
        ],
        "aborted_transactions": {}
      }
    }
  }
}

Copied!

Each entry in records carries the message value, key, timestamp, and offset. When config.message_deserializer is json, value is a parsed object. When it’s noop, value is the raw message as a string.

Which records a request returns depends on config.auto_offset_reset. When it’s set to latest (default), it returns only messages produced after the Policy started consuming. earliest returns messages from the beginning of the topic.

Responses aren’t immediate. A request can take several seconds to return, even when records are already available on the topic.

Message delivery guarantees

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

Schema registry support

You can integrate the Kafka Consume Policy 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 Policy to validate and serialize/deserialize messages in a standardized format.

Using a schema registry with AI 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:

How schema registry validation works

When a consume policy is configured with a schema registry, the following workflow occurs:

 
sequenceDiagram
autonumber
    participant Kafka
    participant Kong as Kafka Consume Policy
    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 Policy, use the config.schema_registry parameter for a Policy-wide registry, or config.topics[].schema_registry to override it for an individual topic.

Filter and transform messages

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

Authentication

The Kafka Consume Policy supports the following SASL authentication mechanisms for broker connections through config.authentication.mechanism:

Mechanism	Description
`PLAIN`	Authenticates using a username and password. Set `authentication.strategy` to `sasl` and provide `authentication.user` and `authentication.password`.
`SCRAM-SHA-256`	Authenticates using a username and password with SCRAM-SHA-256 hashing. Set `authentication.strategy` to `sasl` and provide `authentication.user` and `authentication.password`.
`SCRAM-SHA-512`	Authenticates using a username and password with SCRAM-SHA-512 hashing. Set `authentication.strategy` to `sasl` and provide `authentication.user` and `authentication.password`.

Configure TLS for broker connections with config.security.

Example

The following example creates a Kafka Consume Policy that reads from two topics over HTTP GET, authenticating to the brokers with SASL/PLAIN. Reference it from the policies array on an entity that defines config.route to make it reachable, as described in Consume messages:

policy.yaml