Create a how-to for using the Kafka Upstream policy

Start by reading app/_ai_gateway_policies/kafka-upstream/index.md and reviewing the supplied local Kafka setup and request flow. Restructure the material as a clear how-to with prerequisites, commands, verification steps, and expected output. Done means a reader can run the example locally, send a request, and confirm the message in the kong-upstream topic.

Transform this to a proper how-to:

Testing the Kafka Upstream Policy example locally

This is the minimum setup needed to run the example in
app/_ai_gateway_policies/kafka-upstream/index.md against a local Kafka broker.

Requires kafkactl, jq, and the usual AI Gateway prereqs.

Working directory and SASL credentials

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

Broker

Single-node KRaft broker with two listeners. SASL on 9092 advertised as host.docker.internal
for the data plane, plaintext on 9094 advertised as localhost for local tooling.
host.docker.internal doesn't resolve from the host itself, which is why both exist.

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 before involving the gateway

Run this from a separate container, not from inside the kafka container. It exercises the same
network path the data plane uses.

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

You should get a list of broker API versions ending in ), with SaslHandshake(17) and
SaslAuthenticate(36) present and no SaslAuthenticationException.

Create the topic

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

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

The Policy, plus an AI Agent to accept requests

Unlike Kafka Log, this Policy terminates the request: it publishes to Kafka and returns without
proxying. The AI Agent exists only to supply a Route, so config.url is never reached.

ssl is set to false here to simplify testing.

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"

Send requests, then read the topic

Allow a few seconds after kongctl sync for the data plane to pick up the new configuration.
Requests sent before it does return 404.

Send a request with a JSON body:

curl -i -X POST "$KONNECT_PROXY_URL/upstream-test" \
  -H 'Content-Type: application/json' \
  -d '{"hello":"world"}'

The Policy terminates the request, so expect a 200 with an acknowledgement rather than a
response from config.url. A 502 means the Policy didn't engage and the request went to the dead
upstream port instead.

Read the message back:

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

You should see the following output:

{
  "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": {}
}