Consume Kafka messages over HTTP with the Kafka Consume Policy

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

mkdir -p ~/kafka-consume-test && cd ~/kafka-consume-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

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

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 a kafkactl context that connects through the plaintext listener, then create the ai-events and ai-audit topics:

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

kafkactl -C kafkactl.yaml --context direct create topic ai-events

Create the second topic:

kafkactl -C kafkactl.yaml --context direct create topic ai-audit

Create a Kafka Consume Policy and attach it to an AI Agent that supplies the Route the Policy serves from.

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-consume.yaml
_defaults:
  kongctl:
    namespace: kafka-test

ai_gateway_policies:
  - ref: kafka-consume
    ai_gateway: !lookup {id: !env AI_GATEWAY_ID}
    name: kafka-consume
    display_name: Kafka Consume
    type: kafka-consume
    enabled: true
    global: false
    config:
      bootstrap_servers:
        - host: host.docker.internal
          port: 9092
      topics:
        - name: ai-events
        - name: ai-audit
      mode: http-get
      message_deserializer: json
      auto_offset_reset: latest
      enforce_latest_offset_reset: true
      authentication:
        strategy: sasl
        mechanism: PLAIN
        user: kafka_user
        password: kafka-password
      security:
        ssl: false

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

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

The Policy uses auto_offset_reset: latest, so the consumer starts at the end of each topic and only returns messages produced after it starts consuming. Prime the consumer first, then produce messages.

1. Send a first request to start the consumer. Requests sent before the data plane picks up the new configuration return 404, so wait a few seconds after kongctl sync. The response contains no records yet:

   curl -s "$KONNECT_PROXY_URL/kafka" | jq .

   Copied!

2. Produce a message to the ai-events topic:

   kafkactl -C kafkactl.yaml --context direct produce ai-events --value='{"event":"prompt_received","model":"gpt-4o"}'

   Copied!

3. Produce a message to the ai-audit topic:

   kafkactl -C kafkactl.yaml --context direct produce ai-audit --value='{"actor":"user","action":"policy_applied"}'

   Copied!

4. Send a second request to read the messages:

   curl -s "$KONNECT_PROXY_URL/kafka" | jq .

   Copied!

   The response groups records by topic name, 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": {}
         }
       }
     },
     "ai-audit": {
       "partitions": {
         "0": {
           "high_watermark": 1,
           "last_stable_offset": 1,
           "errcode": 0,
           "records": [
             {
               "value": {"actor": "user", "action": "policy_applied"},
               "key": "",
               "timestamp": 1787305085290,
               "offset": 0
             }
           ],
           "aborted_transactions": {}
         }
       }
     }
   }