Create a how-to for using the Kafka Consume policy

Transform this to a proper how-to:

# Testing the Kafka Consume Policy example locally

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

Requires [kafkactl](https://github.com/deviceinsight/kafkactl?tab=readme-ov-file#installation), `jq`, and the usual AI Gateway prereqs.

## Working directory and SASL credentials

```bash
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
```

## 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.

```bash
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.

```bash
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 something like this:

```sh
host.docker.internal:9092 (id: 0 rack: null isFenced: false) -> (
Produce(0): 0 to 13 [usable: 13],
Fetch(1): 4 to 18 [usable: 18],
ListOffsets(2): 1 to 11 [usable: 11],
Metadata(3): 0 to 13 [usable: 13],
OffsetCommit(8): 2 to 10 [usable: 10],
OffsetFetch(9): 1 to 10 [usable: 10],
FindCoordinator(10): 0 to 6 [usable: 6],
JoinGroup(11): 0 to 9 [usable: 9],
Heartbeat(12): 0 to 4 [usable: 4],
LeaveGroup(13): 0 to 5 [usable: 5],
SyncGroup(14): 0 to 5 [usable: 5],
DescribeGroups(15): 0 to 6 [usable: 6],
ListGroups(16): 0 to 5 [usable: 5],
SaslHandshake(17): 0 to 1 [usable: 1],
ApiVersions(18): 0 to 4 [usable: 4],
CreateTopics(19): 2 to 7 [usable: 7],
DeleteTopics(20): 1 to 6 [usable: 6],
DeleteRecords(21): 0 to 2 [usable: 2],
InitProducerId(22): 0 to 5 [usable: 5],
OffsetForLeaderEpoch(23): 2 to 4 [usable: 4],
AddPartitionsToTxn(24): 0 to 5 [usable: 5],
AddOffsetsToTxn(25): 0 to 4 [usable: 4],
EndTxn(26): 0 to 5 [usable: 5],
WriteTxnMarkers(27): 1 to 2 [usable: 2],
TxnOffsetCommit(28): 0 to 5 [usable: 5],
DescribeAcls(29): 1 to 3 [usable: 3],
CreateAcls(30): 1 to 3 [usable: 3],
DeleteAcls(31): 1 to 3 [usable: 3],
DescribeConfigs(32): 1 to 4 [usable: 4],
AlterConfigs(33): 0 to 2 [usable: 2],
AlterReplicaLogDirs(34): 1 to 2 [usable: 2],
DescribeLogDirs(35): 1 to 5 [usable: 5],
SaslAuthenticate(36): 0 to 2 [usable: 2],
CreatePartitions(37): 0 to 3 [usable: 3],
CreateDelegationToken(38): 1 to 3 [usable: 3],
RenewDelegationToken(39): 1 to 2 [usable: 2],
ExpireDelegationToken(40): 1 to 2 [usable: 2],
DescribeDelegationToken(41): 1 to 3 [usable: 3],
DeleteGroups(42): 0 to 2 [usable: 2],
ElectLeaders(43): 0 to 2 [usable: 2],
IncrementalAlterConfigs(44): 0 to 1 [usable: 1],
AlterPartitionReassignments(45): 0 to 1 [usable: 1],
ListPartitionReassignments(46): 0 [usable: 0],
OffsetDelete(47): 0 [usable: 0],
DescribeClientQuotas(48): 0 to 1 [usable: 1],
AlterClientQuotas(49): 0 to 1 [usable: 1],
DescribeUserScramCredentials(50): 0 [usable: 0],
AlterUserScramCredentials(51): 0 [usable: 0],
DescribeQuorum(55): 0 to 2 [usable: 2],
UpdateFeatures(57): 0 to 2 [usable: 2],
DescribeCluster(60): 0 to 2 [usable: 2],
DescribeProducers(61): 0 [usable: 0],
UnregisterBroker(64): 0 [usable: 0],
DescribeTransactions(65): 0 [usable: 0],
ListTransactions(66): 0 to 2 [usable: 2],
ConsumerGroupHeartbeat(68): 0 to 1 [usable: 1],
ConsumerGroupDescribe(69): 0 to 1 [usable: 1],
GetTelemetrySubscriptions(71): UNSUPPORTED,
PushTelemetry(72): UNSUPPORTED,
ListConfigResources(74): 0 to 1 [usable: 1],
DescribeTopicPartitions(75): 0 [usable: 0],
ShareGroupHeartbeat(76): 1 [usable: 1],
ShareGroupDescribe(77): 1 [usable: 1],
ShareFetch(78): 1 to 2 [usable: 2],
ShareAcknowledge(79): 1 to 2 [usable: 2],
AddRaftVoter(80): 0 to 1 [usable: 1],
RemoveRaftVoter(81): 0 [usable: 0],
InitializeShareGroupState(83): 0 [usable: 0],
ReadShareGroupState(84): 0 [usable: 0],
WriteShareGroupState(85): 0 to 1 [usable: 1],
DeleteShareGroupState(86): 0 [usable: 0],
ReadShareGroupStateSummary(87): 0 to 1 [usable: 1],
StreamsGroupHeartbeat(88): 0 [usable: 0],
StreamsGroupDescribe(89): 0 [usable: 0],
DescribeShareGroupOffsets(90): 0 to 1 [usable: 1],
AlterShareGroupOffsets(91): 0 [usable: 0],
DeleteShareGroupOffsets(92): 0 [usable: 0]
)
```

## Create both topics

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

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

## The Policy, plus an AI Agent to expose it on a Route

The example on its own creates a Policy with nowhere to serve from. The AI Agent supplies the
`config.route` that makes it reachable.

`ssl` is set to `false` here to simplify testing.

```bash
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
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"
```

## Prime the consumer, then produce

The example uses `auto_offset_reset: latest`, so the consumer starts at the end
of each topic and only returns messages produced after it began consuming.

```bash
# First call starts the consumer. Expect empty topics.
curl -s "$KONNECT_PROXY_URL/kafka" | jq .

# Now produce
kafkactl -C kafkactl.yaml --context direct produce ai-events --value='{"event":"prompt_received","model":"gpt-4o"}'
kafkactl -C kafkactl.yaml --context direct produce ai-audit --value='{"actor":"user","action":"policy_applied"}'

# Second call returns them
curl -s "$KONNECT_PROXY_URL/kafka" | jq .
```

Expected response, keyed by topic name and then by partition:

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

Start with app/_ai_gateway_policies/kafka-consume/index.md and compare the supplied setup with the surrounding documentation structure. Verify the local broker, topics, policy, agent, and request flow as described. Done means the page presents a coherent how-to from prerequisites through the expected Kafka response.

Written by the indexing model from the issue text.