Request Rule Validator: Enforce topic naming, sizing, and retention conventions - Policy

Enforce topic naming, sizing, and retention conventionsv1.3+

This Request Rule Validator policy enforces topic conventions for a shared Kafka cluster.

In this example:

Topic names must start with team-a..
Topics can’t have more than 50 partitions.
Topics must have a replication factor of at least 3.
retention.ms must be set to 24 hours (86400000).

Clients can change a topic after they create it, so the policy enforces the conventions on more than one request type:

create_topics: Enforces all the conventions when a client creates a topic. If the request uses a manual replica assignment, the partition count and replication factor are -1, so the rules check context.topic.assignments instead.
create_partitions: Prevents clients from growing a topic past 50 partitions.
alter_configs: This request replaces the full configuration of a topic, so it must include retention.ms with the required value. Otherwise, retention.ms goes back to the broker default.
incremental_alter_configs: Clients can change retention.ms only with the SET operation and the required value. This rule also prevents clients from deleting retention.ms.

Each rule’s expression describes the request state that’s allowed. When an item in the request, for example a topic, doesn’t match a rule, Event Gateway rejects only that item with the Kafka POLICY_VIOLATION error code. Other items in the same batched request are still processed.

Set up the policy

curl -X POST https://{region}.api.konghq.com/v1/event-gateways/{eventGatewayId}/virtual-clusters/{virtualClusterId}/cluster-policies \
    --header "accept: application/json" \
    --header "Content-Type: application/json" \
    --header "Authorization: Bearer $KONNECT_TOKEN" \
    --data '
    {
      "name": "request_rule_validator_policy",
      "type": "request_rule_validator",
      "config": {
        "requests": [
          {
            "type": "create_topics",
            "rules": [
              {
                "description": "Topic name must start with \"team-a.\"",
                "rule": "{{context.topic.name.startsWith(\"team-a.\")}}",
                "action": "reject"
              },
              {
                "description": "Topic must have 50 partitions or fewer",
                "rule": "{{context.topic.num_partitions != -1 ? context.topic.num_partitions <= 50 : size(context.topic.assignments) <= 50}}",
                "action": "reject"
              },
              {
                "description": "Replication factor must be at least 3",
                "rule": "{{context.topic.replication_factor != -1 ? context.topic.replication_factor >= 3 : context.topic.assignments.all(a, size(a.broker_ids) >= 3)}}",
                "action": "reject"
              },
              {
                "description": "retention.ms must be set to 24h (86400000)",
                "rule": "{{\"retention.ms\" in context.topic.configs && int(context.topic.configs[\"retention.ms\"]) == 86400000}}",
                "action": "reject"
              }
            ]
          },
          {
            "type": "create_partitions",
            "rules": [
              {
                "description": "Topic must have 50 partitions or fewer",
                "rule": "{{context.topic.count <= 50}}",
                "action": "reject"
              }
            ]
          },
          {
            "type": "alter_configs",
            "rules": [
              {
                "description": "retention.ms must be set to 24h (86400000)",
                "rule": "{{context.resource.type != \"TOPIC\" || (\"retention.ms\" in context.resource.configs && int(context.resource.configs[\"retention.ms\"]) == 86400000)}}",
                "action": "reject"
              }
            ]
          },
          {
            "type": "incremental_alter_configs",
            "rules": [
              {
                "description": "retention.ms can only be set to 24h (86400000)",
                "rule": "{{context.resource.type != \"TOPIC\" || context.config.name != \"retention.ms\" || (context.config.operation == \"SET\" && int(context.config.value) == 86400000)}}",
                "action": "reject"
              }
            ]
          }
        ]
      }
    }
    '

Copied!

Make sure to replace the following placeholders with your own values:

region: Geographic region where your Kong Konnect is hosted and operates.
KONNECT_TOKEN: Your Personal Access Token (PAT) associated with your Konnect account.
virtualClusterId: The id of the Virtual Cluster.
eventGatewayId: The id of the Event Gateway.
eventGatewayListenerId: The id of the Event Gateway Listener.

See the Konnect Event Gateway API reference to learn about region-specific URLs and personal access tokens.

Prerequisite: Configure your Personal Access Token

terraform {
  required_providers {
    konnect-beta = {
      source  = "kong/konnect-beta"
    }
  }
}

provider "konnect-beta" {
  personal_access_token = "$KONNECT_TOKEN"
  server_url            = "https://us.api.konghq.com/"
}

Copied!

Add the following to your Terraform configuration:

resource "konnect_event_gateway_cluster_policy_request_rule_validator" "my_virtual_cluster_policy_request_rule_validator" {
  provider = konnect-beta
  type = "request_rule_validator"
  config = {
    requests = [
      {
        type = "create_topics"
        rules = [
          {
            description = "Topic name must start with \"team-a.\""
            rule = "{{context.topic.name.startsWith(\"team-a.\")}}"
            action = "reject"
          }, 

          {
            description = "Topic must have 50 partitions or fewer"
            rule = "{{context.topic.num_partitions != -1 ? context.topic.num_partitions <= 50 : size(context.topic.assignments) <= 50}}"
            action = "reject"
          }, 

          {
            description = "Replication factor must be at least 3"
            rule = "{{context.topic.replication_factor != -1 ? context.topic.replication_factor >= 3 : context.topic.assignments.all(a, size(a.broker_ids) >= 3)}}"
            action = "reject"
          }, 

          {
            description = "retention.ms must be set to 24h (86400000)"
            rule = "{{\"retention.ms\" in context.topic.configs && int(context.topic.configs[\"retention.ms\"]) == 86400000}}"
            action = "reject"
          }        ]
      }, 

      {
        type = "create_partitions"
        rules = [
          {
            description = "Topic must have 50 partitions or fewer"
            rule = "{{context.topic.count <= 50}}"
            action = "reject"
          }        ]
      }, 

      {
        type = "alter_configs"
        rules = [
          {
            description = "retention.ms must be set to 24h (86400000)"
            rule = "{{context.resource.type != \"TOPIC\" || (\"retention.ms\" in context.resource.configs && int(context.resource.configs[\"retention.ms\"]) == 86400000)}}"
            action = "reject"
          }        ]
      }, 

      {
        type = "incremental_alter_configs"
        rules = [
          {
            description = "retention.ms can only be set to 24h (86400000)"
            rule = "{{context.resource.type != \"TOPIC\" || context.config.name != \"retention.ms\" || (context.config.operation == \"SET\" && int(context.config.value) == 86400000)}}"
            action = "reject"
          }        ]
      }    ]
  }


  virtual_cluster_id = konnect_event_gateway_virtual_cluster.my_virtual_cluster.id

  gateway_id = konnect_event_gateway.my_event_gateway.id
}

Copied!

The following example creates a new request_rule_validator policy.

kongctl

event_gateway_policy.yaml

Copied!

event_gateways:
  - ref: eventGatewayName
    name: eventGatewayName
    virtual_clusters:
    - ref: virtualClusterName
      name: virtualClusterName
      cluster_policies:
      - ref: request_rule_validator_policy
        type: request_rule_validator
        request_rule_validator:
          name: request_rule_validator_policy
          config:
            requests:
            - type: create_topics
              rules:
              - description: Topic name must start with "team-a."
                rule: '{{context.topic.name.startsWith("team-a.")}}'
                action: reject
              - description: Topic must have 50 partitions or fewer
                rule: "{{context.topic.num_partitions != -1 ? context.topic.num_partitions
                  <= 50 : size(context.topic.assignments) <= 50}}"
                action: reject
              - description: Replication factor must be at least 3
                rule: "{{context.topic.replication_factor != -1 ? context.topic.replication_factor
                  >= 3 : context.topic.assignments.all(a, size(a.broker_ids) >= 3)}}"
                action: reject
              - description: retention.ms must be set to 24h (86400000)
                rule: '{{"retention.ms" in context.topic.configs && int(context.topic.configs["retention.ms"])
                  == 86400000}}'
                action: reject
            - type: create_partitions
              rules:
              - description: Topic must have 50 partitions or fewer
                rule: "{{context.topic.count <= 50}}"
                action: reject
            - type: alter_configs
              rules:
              - description: retention.ms must be set to 24h (86400000)
                rule: '{{context.resource.type != "TOPIC" || ("retention.ms" in context.resource.configs
                  && int(context.resource.configs["retention.ms"]) == 86400000)}}'
                action: reject
            - type: incremental_alter_configs
              rules:
              - description: retention.ms can only be set to 24h (86400000)
                rule: '{{context.resource.type != "TOPIC" || context.config.name !=
                  "retention.ms" || (context.config.operation == "SET" && int(context.config.value)
                  == 86400000)}}'
                action: reject

Make sure to replace the following placeholders with your own values:

eventGatewayName: The name of your Event Gateway.
virtualClusterName: The name of the Virtual Cluster.

Request Rule Validator

Enforce topic naming, sizing, and retention conventionsv1.3+

Set up the policy

Help us make these docs great!

Still need help?