Home / Event Gateway
Edit
Report

Enforce rules on Kafka requests with Kong Event Gateway

Uses: Event Gateway Konnect API
Incompatible with
on-prem
Related Documentation
Event Gateway Documentation
Tags
#event-gateway #kafka
Related Resources
Request Rule Validator policy
Expressions reference
Virtual clusters
Kafka ACL policy
Event Gateway Control Plane API
Minimum Version
Event Gateway - 1.3
TL;DR
  1. Create a Request Rule Validator cluster policy on your virtual cluster.
  2. Add rules that describe the valid state of a request, for example a topic naming convention or a partition limit.
  3. Try to break the rules. Kong Event Gateway rejects the requests.

Prerequisites

Install kafkactl

Install kafkactl. You’ll need it to interact with Kafka clusters.

Start a local Kafka cluster

Start a Docker Compose cluster with multiple Kafka services.

First, we need to create a docker-compose.yaml file. This file will define the services we want to run in our local environment:

cat <<EOF > docker-compose.yaml
name: kafka_cluster

networks:
  kafka:
    name: kafka_event_gateway

services:
  kafka1:
    image: apache/kafka:4.3.1
    networks:
      - kafka
    container_name: kafka1
    ports:
      - "9094:9094"
    environment:
      KAFKA_NODE_ID: 0
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: INTERNAL://kafka1:9092,CONTROLLER://kafka1:9093,EXTERNAL://0.0.0.0:9094
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:9092,EXTERNAL://localhost:9094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_CONTROLLER_QUORUM_VOTERS: 0@kafka1:9093,1@kafka2:9093,2@kafka3:9093
      KAFKA_CLUSTER_ID: 'abcdefghijklmnopqrstuv'
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs

  kafka2:
    image: apache/kafka:4.3.1
    networks:
      - kafka
    container_name: kafka2
    ports:
      - "9095:9095"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: INTERNAL://kafka2:9092,CONTROLLER://kafka2:9093,EXTERNAL://0.0.0.0:9095
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka2:9092,EXTERNAL://localhost:9095
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_CONTROLLER_QUORUM_VOTERS: 0@kafka1:9093,1@kafka2:9093,2@kafka3:9093
      KAFKA_CLUSTER_ID: 'abcdefghijklmnopqrstuv'
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs

  kafka3:
    image: apache/kafka:4.3.1
    networks:
      - kafka
    container_name: kafka3
    ports:
      - "9096:9096"
    environment:
      KAFKA_NODE_ID: 2
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: INTERNAL://kafka3:9092,CONTROLLER://kafka3:9093,EXTERNAL://0.0.0.0:9096
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka3:9092,EXTERNAL://localhost:9096
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_CONTROLLER_QUORUM_VOTERS: 0@kafka1:9093,1@kafka2:9093,2@kafka3:9093
      KAFKA_CLUSTER_ID: 'abcdefghijklmnopqrstuv'
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs

  schema-registry:
    image: confluentinc/cp-schema-registry:8.3.2
    networks:
      - kafka
    container_name: schema-registry
    depends_on:
      - kafka1
      - kafka2
      - kafka3
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka1:9092,kafka2:9092,kafka3:9092

EOF

Now, let’s start the local setup:

docker compose up -d

Kong Konnect

If you don’t have a Konnect account, you can get started quickly with our onboarding wizard.

  1. The following Konnect items are required to complete this tutorial:
    • Personal access token (PAT): Create a new personal access token by opening the Konnect PAT page and selecting Generate Token.

  2. Set the personal access token as an environment variable:

    export KONNECT_TOKEN='YOUR KONNECT TOKEN'

Kong Event Gateway running

Run the quickstart script to automatically provision a demo Kong Gateway control plane and data plane, and configure your environment:

curl -Ls https://get.konghq.com/event-gateway | bash -s -- -k $KONNECT_TOKEN -N kafka_event_gateway

This sets up an Kong Gateway control plane named event-gateway-quickstart, provisions a local data plane, and prints out the following environment variable export:

export EVENT_GATEWAY_ID=your-gateway-id

Copy and paste the command with your Event Gateway ID into your terminal to configure your session.

This quickstart script is meant for demo purposes only, therefore it runs locally with most default parameters and a small number of exposed ports. If you want to run Kong Gateway as a part of a production-ready platform, set up your control plane and data planes through the Konnect UI, or using Terraform.

Overview

In this guide, you’ll learn how to reject Kafka requests that break your platform conventions.

Kafka ACLs control who can do what, but they can’t read the content of a request. They can say who may create topics, but not which names or partition counts are allowed. The Request Rule Validator policy checks the content of each request. Every rule is a boolean expression over the request fields, and Kong Event Gateway rejects requests that break a rule.

We’ll act as a platform team that runs a shared Kafka cluster. Teams create topics and produce records through a virtual cluster, and the platform enforces three rules:

Kafka request

Rule
CreateTopics Topic names must start with the team prefix team-a.
CreateTopics A topic can have at most 5 partitions
Produce Producers must require acknowledgements from all in-sync replicas, so no record is lost if a broker fails

The Request Rule Validator is a cluster policy, which means it runs on all Kafka API commands that pass through the virtual cluster, before the requests reach the backend cluster:

 
flowchart LR
    P[kafkactl] --> RR

    subgraph GW [Event Gateway virtual cluster]
        RR[Request Rule
Validator]
    end

    RR -->|valid| K[Kafka
broker]
    RR -.->|rule violated| X[Request
rejected]

Every rule carries a description. When a request violates the rule, Kong Event Gateway returns this text in the error, so clients learn why the request was rejected.

Create a backend cluster

Use the following command to create a backend cluster that connects to the Kafka servers you set up:

BACKEND_CLUSTER_ID=$(curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/backend-clusters" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "name": "backend_cluster",
       "bootstrap_servers": [
         "kafka1:9092",
         "kafka2:9092",
         "kafka3:9092"
       ],
       "authentication": {
         "type": "anonymous"
       },
       "tls": {
         "enabled": false
       },
       "insecure_allow_anonymous_virtual_cluster_auth": true
     }' | jq -r ".id"
)

Create a virtual cluster

Create a virtual cluster that clients connect to:

VIRTUAL_CLUSTER_ID=$(curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/virtual-clusters" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "name": "orders_vc",
       "destination": {
         "id": "'$BACKEND_CLUSTER_ID'"
       },
       "dns_label": "orders",
       "authentication": [
         {
           "type": "anonymous"
         }
       ],
       "acl_mode": "passthrough"
     }' | jq -r ".id"
)

Create a listener with a forwarding policy

Create a listener to accept connections:

LISTENER_ID=$(curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/listeners" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "name": "orders_listener",
       "addresses": [
         "0.0.0.0"
       ],
       "ports": [
         "19092-19095"
       ]
     }' | jq -r ".id"
)

Create a Forward to Virtual Cluster policy to forward traffic to the virtual cluster:

curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/listeners/$LISTENER_ID/policies" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "type": "forward_to_virtual_cluster",
       "name": "forward_to_orders_vc",
       "config": {
         "type": "port_mapping",
         "advertised_host": "localhost",
         "destination": {
           "id": "'$VIRTUAL_CLUSTER_ID'"
         }
       }
     }'

For demo purposes, we’re using port mapping, which assigns each Kafka broker to a dedicated port on the Event Gateway. In production, we recommend using SNI routing instead.

Create the Request Rule Validator policy

Create a Request Rule Validator cluster policy on the virtual cluster. The requests array groups rules by Kafka request type, and this guide configures two types:

POLICY_ID=$(curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/virtual-clusters/$VIRTUAL_CLUSTER_ID/cluster-policies" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "type": "request_rule_validator",
       "name": "topic_governance",
       "config": {
         "requests": [
           {
             "type": "create_topics",
             "rules": [
               {
                 "description": "Topic names must start with the team prefix team-a.",
                 "rule": "context.topic.name.startsWith(\"team-a.\")",
                 "action": "reject"
               },
               {
                 "description": "Topics can have at most 5 partitions",
                 "rule": "context.topic.num_partitions <= 5",
                 "action": "reject"
               }
             ]
           },
           {
             "type": "produce",
             "rules": [
               {
                 "description": "Producers must require acknowledgements from all in-sync replicas",
                 "rule": "context.produce.acks == -1",
                 "action": "reject"
               }
             ]
           }
         ]
       }
     }' | jq -r ".id"
)

In this configuration:

  • Each rule is a boolean expression that describes the valid state of a request. The action runs when the expression evaluates to false. An expression that can’t be evaluated, for example because a value has an unexpected format, also counts as false.
  • context.topic.name.startsWith("team-a.") and context.topic.num_partitions <= 5 are evaluated against each topic in a CreateTopics request. A single request can create several topics, and each topic is checked independently.
  • context.produce.acks == -1 is evaluated against every Produce request. The value -1 means the producer waits for acknowledgements from all in-sync replicas.
  • action: reject fails the request with the POLICY_VIOLATION error code. The alternative is passthrough, which lets the request continue and only logs the violation. Use passthrough to test a rule before you start enforcing it.

Configure kafkactl

Create a kafkactl configuration with a vc context that connects through the virtual cluster:

cat <<EOF > kafkactl.yaml
contexts:
  vc:
    brokers:
      - localhost:19092
EOF

Create a topic

Create the team-a.orders topic with 3 partitions. The request matches both create_topics rules, so the Event Gateway forwards it to the backend cluster:

kafkactl -C kafkactl.yaml --context vc create topic team-a.orders --partitions 3

kafkactl prints the name of the created topic:

topic created: team-a.orders

Validate

Violate the naming rule

Try to create a topic whose name doesn’t start with team-a.:

kafkactl -C kafkactl.yaml --context vc create topic team-b.orders --partitions 3

The request fails with the POLICY_VIOLATION error code, and the error contains the rule description:

failed to create topic: kafka server: Request parameters do not satisfy the configured policy - Topic names must start with the team prefix team-a.

Violate the partition limit

Try to create a topic with more than 5 partitions:

kafkactl -C kafkactl.yaml --context vc create topic team-a.big --partitions 50
failed to create topic: kafka server: Request parameters do not satisfy the configured policy - Topics can have at most 5 partitions

Reject records produced without full acknowledgement

Produce a record and require an acknowledgement from the leader only, which violates the produce rule:

echo '{"order":{"id":"1","item":"coffee","qty":2}}' | kafkactl -C kafkactl.yaml --context vc produce team-a.orders --required-acks WaitForLocal

kafkactl reports that no message was produced and exits with code 1. The produce request was rejected with the POLICY_VIOLATION error code, so the record never reaches the backend cluster:

0 messages produced
Failed to produce message: kafka server: Request parameters do not satisfy the configured policy

Unlike topic creation, the error doesn’t contain the rule description, because the Produce response doesn’t carry it.

Produce with full acknowledgement

Produce a different record with --required-acks WaitForAll, which maps to the value -1 that the rule requires:

echo '{"order":{"id":"2","item":"tea","qty":1}}' | kafkactl -C kafkactl.yaml --context vc produce team-a.orders --required-acks WaitForAll

Consume the topic to confirm that only the compliant record was stored:

kafkactl -C kafkactl.yaml --context vc consume team-a.orders --from-beginning --exit
{"order":{"id":"2","item":"tea","qty":1}}

Cleanup

Clean up Kong Event Gateway resources

When you’re done experimenting with this example, clean up the resources:

  1. If you created a new Event Gateway control plane and want to conserve your free trial credits or avoid unnecessary charges, delete the new control plane used in this tutorial.

  2. Stop and remove the containers:

    docker-compose down

This will stop all services and remove the containers, but preserve your configuration files for future use.

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