Skip to main content

kafka-logger

The kafka-logger plugin sends request and response logs as JSON objects to Apache Kafka in batches and supports customizable log formats.

Examples​

The examples show how to send gateway request logs to Kafka, customize their contents, and secure the broker connection with TLS.

The examples use the official Apache Kafka 3.9.2 image in single-node KRaft mode.

Kafka compatibility

kafka-logger supports Kafka Produce API versions 0–2. Kafka 4 removes those versions, so use a Kafka 3.x broker until the plugin supports Produce API version 3 or later.

Set GATEWAY_CONTAINER to the running APISIX or API7 Gateway container:

export GATEWAY_CONTAINER=replace-with-gateway-container-name

Create a dedicated network for the gateway and Kafka:

docker network create gateway-kafka-net

Connect the gateway to the network:

docker network connect gateway-kafka-net "$GATEWAY_CONTAINER"

Create the following Docker Compose file:

docker-compose.yml
services:
kafka-server:
image: apache/kafka:3.9.2
container_name: kafka-server
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-server:9092
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-server:9093
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
networks:
- kafka

networks:
kafka:
name: gateway-kafka-net
external: true

Start the broker:

docker compose up -d

After the container starts, verify that the broker accepts requests:

docker exec kafka-server /opt/kafka/bin/kafka-topics.sh \
--bootstrap-server localhost:9092 \
--list

If the command reports a connection error, wait a few seconds and run it again.

Create the topic used by the examples:

docker exec kafka-server /opt/kafka/bin/kafka-topics.sh \
--bootstrap-server localhost:9092 \
--create \
--if-not-exists \
--topic apisix-logs

To inspect records produced by the examples, run a consumer in a separate terminal:

docker exec -it kafka-server /opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server localhost:9092 \
--topic apisix-logs \
--from-beginning

Log in Different Meta Log Formats​

The following example sends route request logs to Kafka and shows the difference between the default and origin meta formats.

Create a route with kafka-logger as follows:

curl "http://127.0.0.1:9180/apisix/admin/routes" -X PUT \
-H "X-API-KEY: ${ADMIN_API_KEY}" \
-d '{
"id": "kafka-logger-route",
"uri": "/get",
"plugins": {
"kafka-logger": {
"meta_format": "default",
"brokers": [
{
"host": "kafka-server",
"port": 9092
}
],
"kafka_topic": "apisix-logs",
"key": "key1",
"batch_max_size": 1
}
},
"upstream": {
"nodes": {
"httpbin.org:80": 1
},
"type": "roundrobin"
}
}'

❶ meta_format: set to the default log format.

❷ batch_max_size: set to 1 to send the log entry immediately.

Send a request to the route to generate a log entry:

curl -i "http://127.0.0.1:9080/get"

You should see an HTTP/1.1 200 OK response.

You should see a log entry in the Kafka topic similar to the following. Addresses, timing values, and version details vary by environment:

{
"latency": 1030.9998989105,
"request": {
"querystring": {},
"headers": {
"host": "127.0.0.1:9080",
"user-agent": "curl/8.7.1",
"accept": "*/*",
"x-forwarded-proto": "http",
"x-forwarded-host": "127.0.0.1:9080",
"x-forwarded-port": "9080"
},
"method": "GET",
"size": 80,
"uri": "/get",
"url": "http://127.0.0.1:9080/get"
},
"response": {
"headers": {
"content-length": "311",
"access-control-allow-credentials": "true",
"content-type": "application/json",
"connection": "close",
"access-control-allow-origin": "*",
"date": "Fri, 18 Sep 2026 03:32:31 GMT",
"server": "APISIX/3.18.0"
},
"status": 200,
"size": 539
},
"route_id": "kafka-logger-route",
"client_ip": "192.168.155.1",
"server": {
"hostname": "dd2886d0b7bf",
"version": "3.18.0"
},
"apisix_latency": 408.99989891052,
"service_id": "",
"upstream_latency": 622,
"start_time": 1789702349311,
"upstream": "98.88.64.13:80"
}

Update the meta log format to origin:

curl "http://127.0.0.1:9180/apisix/admin/routes/kafka-logger-route" -X PATCH \
-H "X-API-KEY: ${ADMIN_API_KEY}" \
-d '{
"plugins": {
"kafka-logger": {
"meta_format": "origin"
}
}
}'

Send a request to the route again to generate a new log entry:

curl -i "http://127.0.0.1:9080/get"

You should see an HTTP/1.1 200 OK response.

You should see a log entry in the Kafka topic similar to the following:

GET /get HTTP/1.1
x-forwarded-proto: http
x-forwarded-host: 127.0.0.1
user-agent: curl/8.7.1
x-forwarded-port: 9080
host: 127.0.0.1:9080
accept: */*

Send Logs to a TLS-Enabled Broker​

The following Docker example starts a local Kafka 3.9.2 broker with a CA-signed TLS certificate on gateway-kafka-net. It then connects with certificate verification and uses Produce API version 2 so Kafka records the message timestamp. Install OpenSSL and the Java keytool command before continuing.

Generate a sample CA, a broker certificate for the kafka-tls container hostname, and the Java key store and trust store files required by Kafka:

mkdir -p kafka-tls-certs

openssl req -x509 -newkey rsa:2048 -nodes -days 365 \
-subj "/CN=kafka-example-ca" \
-keyout kafka-tls-certs/ca.key \
-out kafka-tls-certs/ca.crt

openssl req -newkey rsa:2048 -nodes \
-subj "/CN=kafka-tls" \
-keyout kafka-tls-certs/server.key \
-out kafka-tls-certs/server.csr
printf "subjectAltName=DNS:kafka-tls\n" > kafka-tls-certs/server-ext.cnf
openssl x509 -req -days 365 \
-in kafka-tls-certs/server.csr \
-CA kafka-tls-certs/ca.crt \
-CAkey kafka-tls-certs/ca.key \
-CAcreateserial \
-extfile kafka-tls-certs/server-ext.cnf \
-out kafka-tls-certs/server.crt

openssl pkcs12 -export \
-name kafka-tls \
-in kafka-tls-certs/server.crt \
-inkey kafka-tls-certs/server.key \
-certfile kafka-tls-certs/ca.crt \
-out kafka-tls-certs/kafka.keystore.p12 \
-passout pass:changeit

keytool -importkeystore -noprompt \
-srckeystore kafka-tls-certs/kafka.keystore.p12 \
-srcstoretype PKCS12 \
-srcstorepass changeit \
-destkeystore kafka-tls-certs/kafka.keystore.jks \
-deststoretype JKS \
-deststorepass changeit \
-destkeypass changeit

keytool -importcert -noprompt \
-alias kafka-example-ca \
-file kafka-tls-certs/ca.crt \
-keystore kafka-tls-certs/kafka.truststore.jks \
-storepass changeit

printf "changeit\n" > kafka-tls-certs/kafka_keystore_creds
printf "changeit\n" > kafka-tls-certs/kafka_ssl_key_creds

Create the client configuration used later to verify the record:

kafka-tls-certs/client.properties
security.protocol=SSL
ssl.truststore.location=/etc/kafka/secrets/kafka.truststore.jks
ssl.truststore.password=changeit
ssl.endpoint.identification.algorithm=https

Start the TLS-enabled broker on the gateway network:

docker run -d \
--name kafka-tls \
--hostname kafka-tls \
--network gateway-kafka-net \
-v "${PWD}/kafka-tls-certs:/etc/kafka/secrets:ro" \
-e KAFKA_NODE_ID=1 \
-e KAFKA_PROCESS_ROLES=broker,controller \
-e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP="SSL:SSL,CONTROLLER:PLAINTEXT" \
-e KAFKA_ADVERTISED_LISTENERS="SSL://kafka-tls:9093" \
-e KAFKA_LISTENERS="SSL://:9093,CONTROLLER://:29093" \
-e KAFKA_CONTROLLER_QUORUM_VOTERS="1@kafka-tls:29093" \
-e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
-e KAFKA_INTER_BROKER_LISTENER_NAME=SSL \
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
-e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
-e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
-e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
-e KAFKA_SSL_KEYSTORE_FILENAME=kafka.keystore.jks \
-e KAFKA_SSL_KEYSTORE_CREDENTIALS=kafka_keystore_creds \
-e KAFKA_SSL_KEY_CREDENTIALS=kafka_ssl_key_creds \
-e KAFKA_SSL_TRUSTSTORE_LOCATION=/etc/kafka/secrets/kafka.truststore.jks \
-e KAFKA_SSL_TRUSTSTORE_PASSWORD=changeit \
-e KAFKA_SSL_CLIENT_AUTH=none \
-e CLUSTER_ID="4L6g3nShT-eMCtK--X86sw" \
apache/kafka:3.9.2

After the container starts, verify that the TLS listener accepts requests:

docker exec kafka-tls /opt/kafka/bin/kafka-topics.sh \
--bootstrap-server kafka-tls:9093 \
--command-config /etc/kafka/secrets/client.properties \
--list

If the command reports a connection error, wait a few seconds and run it again.

Create the topic:

docker exec kafka-tls /opt/kafka/bin/kafka-topics.sh \
--bootstrap-server kafka-tls:9093 \
--command-config /etc/kafka/secrets/client.properties \
--create \
--if-not-exists \
--topic apisix-logs \
--partitions 1 \
--replication-factor 1

Set the broker address and topic for the route configuration:

export KAFKA_TLS_HOST="kafka-tls"
export KAFKA_TLS_PORT="9093"
export KAFKA_TOPIC="apisix-logs"

Copy the generated CA certificate into the gateway container identified by GATEWAY_CONTAINER:

docker cp kafka-tls-certs/ca.crt \
"$GATEWAY_CONTAINER":/usr/local/apisix/conf/kafka-example-ca.crt

Append the certificate to the existing system trust bundle so that the gateway continues to trust the public CA certificates already installed in the container:

docker exec "$GATEWAY_CONTAINER" sh -c '
cat /etc/ssl/certs/ca-certificates.crt \
/usr/local/apisix/conf/kafka-example-ca.crt \
> /usr/local/apisix/conf/combined-ca-bundle.pem
'

Update the trust bundle path in the quickstart configuration and reload the gateway:

docker exec -u 0 "$GATEWAY_CONTAINER" sh -c '
config=/usr/local/apisix/conf/config.yaml
certificate=/usr/local/apisix/conf/combined-ca-bundle.pem
output=$(mktemp /tmp/kafka-config.XXXXXX)

if grep -q "^ ssl_trusted_certificate:" "$config"; then
sed "s#ssl_trusted_certificate:.*#ssl_trusted_certificate: $certificate#" "$config"
elif grep -q "^ ssl:$" "$config"; then
sed "/^ ssl:$/a\\
ssl_trusted_certificate: $certificate" "$config"
else
sed "/^apisix:$/a\\
ssl:\\
ssl_trusted_certificate: $certificate" "$config"
fi > "$output"

cat "$output" > "$config"
rm "$output"
apisix reload
'

For a multi-instance deployment, distribute the combined trust bundle and configuration change to every gateway instance. The kafka-tls hostname matches the DNS name in the sample broker certificate.

Create a route that sends each log entry immediately to the TLS listener:

curl "http://127.0.0.1:9180/apisix/admin/routes" -X PUT \
-H "X-API-KEY: ${ADMIN_API_KEY}" \
-d @- <<EOF
{
"id": "kafka-logger-tls-route",
"uri": "/get",
"plugins": {
"kafka-logger": {
"brokers": [
{
"host": "${KAFKA_TLS_HOST}",
"port": ${KAFKA_TLS_PORT}
}
],
"kafka_topic": "${KAFKA_TOPIC}",
"api_version": 2,
"batch_max_size": 1,
"tls": {
"verify": true
}
}
},
"upstream": {
"nodes": {
"httpbin.org:80": 1
},
"type": "roundrobin"
}
}
EOF

Send a request to generate a log entry:

curl -i "http://127.0.0.1:9080/get"

Consume one record from the broker container and print its timestamp:

docker exec kafka-tls /opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server kafka-tls:9093 \
--topic apisix-logs \
--consumer.config /etc/kafka/secrets/client.properties \
--from-beginning \
--max-messages 1 \
--property print.timestamp=true

The consumed record should contain the request log and a timestamp later than the Unix epoch. If certificate verification fails, confirm that the CA bundle is readable by the gateway and that KAFKA_TLS_HOST matches a name in the broker certificate.

The record should contain at least the following fields, together with a broker timestamp. The default log entry contains additional request, response, latency, and gateway fields.

{
"route_id": "kafka-logger-tls-route",
"request": {
"method": "GET",
"uri": "/get"
},
"response": {
"status": 200
}
}

Add Request and Response Headers With Plugin Metadata​

The following example uses plugin metadata and built-in variables to add selected request and response headers to every kafka-logger instance.

In APISIX, plugin metadata is used to configure the common metadata fields of all plugin instances of the same plugin. It is useful when a plugin is enabled across multiple resources and requires a universal update to their metadata fields.

First, create a route with kafka-logger as follows:

curl "http://127.0.0.1:9180/apisix/admin/routes" -X PUT \
-H "X-API-KEY: ${ADMIN_API_KEY}" \
-d '{
"id": "kafka-logger-route",
"uri": "/get",
"plugins": {
"kafka-logger": {
"meta_format": "default",
"brokers": [
{
"host": "kafka-server",
"port": 9092
}
],
"kafka_topic": "apisix-logs",
"key": "key1",
"batch_max_size": 1
}
},
"upstream": {
"nodes": {
"httpbin.org:80": 1
},
"type": "roundrobin"
}
}'

❶ meta_format: keep the default format so that fields from plugin metadata are included. Fields configured with log_format_extra are ignored when this field is set to origin.

❷ batch_max_size: set to 1 to send the log entry immediately.

Next, configure the plugin metadata for kafka-logger:

curl "http://127.0.0.1:9180/apisix/admin/plugin_metadata/kafka-logger" -X PUT \
-H "X-API-KEY: ${ADMIN_API_KEY}" \
-d '{
"log_format_extra": {
"host": "$host",
"@timestamp": "$time_iso8601",
"client_ip": "$remote_addr",
"env": "$http_env",
"resp_content_type": "$sent_http_Content_Type"
}
}'

❶ log the custom request header env.

❷ log the response header Content-Type.

Send a request to the route with the env header:

curl -i "http://127.0.0.1:9080/get" -H "env: dev"

You should see a log entry in the Kafka topic similar to the following:

{
"@timestamp": "2026-09-18T03:32:31+00:00",
"host": "127.0.0.1",
"client_ip": "192.168.155.1",
"route_id": "kafka-logger-route",
"env": "dev",
"resp_content_type": "application/json"
}

Log Request Bodies Conditionally​

The following example includes request bodies only when a query parameter satisfies a configured expression.

Create a route with kafka-logger as follows:

curl "http://127.0.0.1:9180/apisix/admin/routes" -X PUT \
-H "X-API-KEY: ${ADMIN_API_KEY}" \
-d '{
"id": "kafka-logger-route",
"uri": "/post",
"plugins": {
"kafka-logger": {
"brokers": [
{
"host": "kafka-server",
"port": 9092
}
],
"kafka_topic": "apisix-logs",
"key": "key1",
"batch_max_size": 1,
"include_req_body": true,
"include_req_body_expr": [["arg_log_body", "==", "yes"]]
}
},
"upstream": {
"nodes": {
"httpbin.org:80": 1
},
"type": "roundrobin"
}
}'

❶ include_req_body: set to true to include request body.

❷ include_req_body_expr: only include request body if the URL query string log_body is yes.

Send a request to the route with a URL query string satisfying the condition:

curl -i "http://127.0.0.1:9080/post?log_body=yes" -X POST -d '{"env": "dev"}'

You should see the request body logged:

{
...,
"method": "POST",
"body": "{\"env\": \"dev\"}",
"size": 179
}
}

Send a request to the route without any URL query string:

curl -i "http://127.0.0.1:9080/post" -X POST -d '{"env": "dev"}'

You should not observe the request body in the log.

info

The log_format_extra field shown above preserves the default log entry, including request and response bodies collected by the plugin. If you configure log_format instead, include the corresponding variables explicitly:

{
"include_req_body": true,
"include_resp_body": true,
"log_format": {
"request_body": "$request_body",
"response_body": "$resp_body"
}
}

Body size limits still apply. Use log_format_extra to add custom fields without replacing the default log entry.