Home / How-to Guides
Edit
Report

Publish requests to Kafka with the Kafka Upstream Policy

Uses: AI Gateway kongctl
Incompatible with
on-prem
Related Documentation
AI Gateway Documentation kongctl Documentation
Tags
#ai #kafka #docker
Related Resources
Kafka Upstream Policy
Kafka Log Policy
Kafka Consume Policy
AI Agent entity
AI Policy entity
Minimum Version
AI Gateway - 2.0
TL;DR

Create a Kafka Upstream Policy that points at your Kafka brokers and attach it to an AI Agent. The Policy publishes each request to the configured topic and returns without proxying it upstream.

Prerequisites

AI Gateway running

This is a Konnect tutorial and requires a Konnect personal access token.

  1. Create a new personal access token by opening the Konnect PAT page and selecting Generate Token.

  2. Export your token to an environment variable:

    export KONNECT_TOKEN='YOUR_KONNECT_PAT'

  3. Run the AI Gateway quickstart script to automatically provision a control plane and data plane in Kong Konnect, and configure your environment:

    curl -Ls https://get.konghq.com/ai | bash -s -- -k $KONNECT_TOKEN

This sets up a AI Gateway control plane named ai-quickstart, provisions a local data plane, and prints out the following environment variables export:

export AI_GATEWAY_ID=your-gateway-id
export KONNECT_TOKEN=$KONNECT_TOKEN
export KONNECT_CONTROL_PLANE_NAME=ai-quickstart
export KONNECT_CONTROL_PLANE_URL=https://us.api.konghq.com
export KONNECT_PROXY_URL='http://localhost:8000'

Copy and paste these into your terminal to configure your session.

kongctl   v1.20.2+

This tutorial uses kongctl to manage Konnect resources programmatically. We recommend keeping kongctl up to date with the latest version (1.20.2).

  1. Install kongctl from developer.konghq.com/kongctl.

  2. Verify the installation:

    kongctl version

kafkactl

This tutorial uses kafkactl to create topics and interact with them. Install it:

brew install kafkactl

For other installation methods, see the kafkactl installation instructions.

jq

This tutorial uses jq to format JSON output. Install it:

brew install jq

For other installation methods, see the jq download page.

Create the working directory and SASL credentials

Create a working directory and a JAAS file that defines the SASL/PLAIN user for the broker:

mkdir -p ~/kafka-upstream-test && cd ~/kafka-upstream-test

cat <<'EOF' > kafka_server_jaas.conf
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka_user"
    password="kafka-password"
    user_kafka_user="kafka-password";
};
EOF

Start a Kafka broker

Run a single-node KRaft broker with two listeners:

  • SASL on port 9092, advertised as host.docker.internal. The data plane connects to this listener.
  • Plaintext on port 9094, advertised as localhost. Local tooling connects to this listener, because host.docker.internal doesn’t resolve from the host itself.
cat <<'EOF' > docker-compose.yaml
name: kafka_aigw

services:
  kafka:
    image: apache/kafka:4.3.1
    container_name: kafka
    ports:
      - "9092:9092"
      - "9094:9094"
    environment:
      KAFKA_NODE_ID: 0
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 0@localhost:9093
      KAFKA_LISTENERS: CONTROLLER://localhost:9093,SASL_PLAINTEXT://0.0.0.0:9092,PLAINTEXT://0.0.0.0:9094
      KAFKA_ADVERTISED_LISTENERS: SASL_PLAINTEXT://host.docker.internal:9092,PLAINTEXT://localhost:9094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
      KAFKA_OPTS: -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_CLUSTER_ID: 'abcdefghijklmnopqrstuv'
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
    volumes:
      - ./kafka_server_jaas.conf:/etc/kafka/kafka_server_jaas.conf
EOF

docker compose up -d

Verify SASL authentication

Run the SASL check from a separate container so it uses the same network path as the data plane:

cat <<'EOF' > client.properties
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka_user" password="kafka-password";
EOF

docker run --rm \
  -v "$PWD/client.properties:/tmp/client.properties" \
  apache/kafka:4.3.1 \
  /opt/kafka/bin/kafka-broker-api-versions.sh \
    --bootstrap-server host.docker.internal:9092 \
    --command-config /tmp/client.properties

The output lists the broker API versions, including SaslHandshake(17) and SaslAuthenticate(36), without a SaslAuthenticationException.

Create the Kafka topic

Create a kafkactl context that connects through the plaintext listener, then create the kong-upstream topic:

cat <<'EOF' > kafkactl.yaml
contexts:
  direct:
    brokers:
      - localhost:9094
EOF

kafkactl -C kafkactl.yaml --context direct create topic kong-upstream

Create a Kafka Upstream Policy and an AI Agent

Create a Kafka Upstream Policy and attach it to an AI Agent that supplies the Route. The Policy terminates the request after publishing it to Kafka, so the Agent never forwards the request to its config.url.

This example sets security.ssl to false so the test setup doesn’t need certificates. We don’t recommend this in production because the traffic isn’t encrypted.

cat <<'EOF' > kafka-upstream.yaml
_defaults:
  kongctl:
    namespace: kafka-upstream-test

ai_gateway_policies:
  - ref: kafka-upstream
    ai_gateway: !lookup {id: !env AI_GATEWAY_ID}
    name: kafka-upstream
    display_name: Kafka Upstream
    type: kafka-upstream
    enabled: true
    global: false
    config:
      bootstrap_servers:
        - host: host.docker.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: false

ai_gateway_agents:
  - ref: kafka-upstream-agent
    ai_gateway: !lookup {id: !env AI_GATEWAY_ID}
    name: kafka-upstream-agent
    display_name: "Kafka Upstream Agent"
    type: http
    enabled: true
    policies: [ !ref kafka-upstream#name ]
    config:
      url: http://host.docker.internal:9999
      route:
        paths:
          - /upstream-test
        protocols:
          - http
          - https
        methods:
          - GET
          - POST
        strip_path: true
EOF

kongctl sync -f kafka-upstream.yaml --pat "$KONNECT_TOKEN"

Validate

Wait a few seconds for the data plane to pick up the new configuration. Requests sent before the data plane picks up the configuration return 404.

  1. Send a POST request with a JSON body to the Agent’s Route:
curl -i -X POST "$KONNECT_PROXY_URL/upstream-test" \
     --no-progress-meter --fail-with-body  \
     -H "Content-Type: application/json" \
     --json '{
       "hello": "world"
     }'

The Policy terminates the request, so the response is a 200 with an acknowledgement from the Policy instead of a response from config.url. A 502 means the Policy didn’t engage, so the Agent forwarded the request to the unused upstream port.

  1. Read the message back from the topic:

    kafkactl -C kafkactl.yaml --context direct consume kong-upstream --from-beginning --exit | jq .

    The output contains the request body, headers, method, and URI:

    {
  "body": "{\"hello\":\"world\"}",
  "body_args": {
    "hello": "world"
  },
  "uri": "/upstream-test",
  "method": "POST",
  "headers": {
    "content-type": "application/json",
    "user-agent": "curl/8.7.1",
    "accept": "*/*",
    "host": "localhost:8000",
    "content-length": "17"
  },
  "uri_args": {}
}

Cleanup

Stop the Kafka broker

 
docker compose down

Clean up AI Gateway resources

To clean up all AI Gateway resources created in this guide, run:

curl -Ls https://get.konghq.com/ai | bash -s -- -d
Something wrong?
Report an Issue | Edit this Page
View all Kong documentation

Help us make these docs great!

Kong Developer docs are open source. If you find these useful and want to make them better, contribute today!
Contribute

Still need help?

Ask in our Slack Community
Contact Support