Kafka Upstream Policy

Overview Configuration reference

Implementation details

This Policy uses the lua-resty-kafka client.

Control which parts of the request are included in the message with config.forward_body (enabled by default), config.forward_headers, config.forward_method, and config.forward_uri.

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 Policy 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, the Policy returns an error message and the message isn’t 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 that case, the message has an extra attribute called body_base64 set to true.

Schema registry support

You can integrate the Kafka Upstream 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 producer Policy is configured with a schema registry, the following workflow occurs:

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

Authentication

The Kafka Upstream 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 global Kafka Upstream Policy that publishes each request to the kong-upstream topic, authenticating to the brokers with SASL/PLAIN:

policy.yaml

Copied!

ai_gateway_policies:
  - ref: kafka-upstream
    ai_gateway: !lookup {id: !env AI_GATEWAY_ID}
    display_name: Kafka Upstream
    name: kafka-upstream
    type: kafka-upstream
    enabled: true
    global: true
    config:
      bootstrap_servers:
      - host: broker.internal
        port: 9092
      topic: kong-upstream
      forward_body: true
      forward_headers: true
      forward_method: true
      forward_uri: true
      authentication:
        strategy: sasl
        mechanism: PLAIN
        user: kafka-user
        password: kafka-password
      security:
        ssl: true

Make sure to replace the following placeholders with your own values:

AI_GATEWAY_ID: The id of your AI Gateway.