Request Rule Validator: Enforce producer and consumer settings - Policy

Enforce producer and consumer settingsv1.3+

This Request Rule Validator policy enforces client settings on Produce and Fetch requests.

In this example, for produce requests:

Producers must set acks to all (-1), so that every in-sync replica stores a record before the broker acknowledges it.
If a producer uses transactions, its transactional ID must start with team-a..

For fetch requests:

Consumers must use the read_committed isolation level, so that they don’t read records from aborted transactions.
Consumers must set fetch.max.wait.ms to at least 100 ms, so that the broker can hold a fetch until data is available instead of returning empty responses in a tight loop.
A fetch response can be at most 50 MiB (fetch.max.bytes).

The default fetch settings of most Kafka clients match these limits: consumers set fetch.max.wait.ms to 500 ms and fetch at most 50 MiB. Most clients use read_uncommitted by default, so consumers must set isolation.level=read_committed explicitly. A consumer that breaks a fetch rule doesn’t receive any records. For example, the Java consumer stops with IllegalStateException: Unexpected error code 44.

When a request doesn’t match a rule, Event Gateway rejects the entire request with the Kafka POLICY_VIOLATION error code. The Produce and Fetch responses don’t carry an error message, so clients only receive the error code, not the rule description.

A producer with acks=0 doesn’t wait for a response. If Event Gateway rejects its records, the producer doesn’t receive an error, and the records are lost.

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": "produce_fetch_settings_policy",
      "type": "request_rule_validator",
      "config": {
        "requests": [
          {
            "type": "produce",
            "rules": [
              {
                "description": "Producers must use acks=all",
                "rule": "{{context.produce.acks == -1}}",
                "action": "reject"
              },
              {
                "description": "Transactional ID must start with \"team-a.\"",
                "rule": "{{dyn(context.produce.transactional_id) == null || context.produce.transactional_id.startsWith(\"team-a.\")}}",
                "action": "reject"
              }
            ]
          },
          {
            "type": "fetch",
            "rules": [
              {
                "description": "Consumers must use the read_committed isolation level",
                "rule": "{{context.fetch.isolation_level == \"read_committed\"}}",
                "action": "reject"
              },
              {
                "description": "fetch.max.wait.ms must be at least 100 ms",
                "rule": "{{context.fetch.max_wait_ms >= 100}}",
                "action": "reject"
              },
              {
                "description": "Fetch responses can be at most 50 MiB",
                "rule": "{{context.fetch.max_bytes <= 52428800}}",
                "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 = "produce"
        rules = [
          {
            description = "Producers must use acks=all"
            rule = "{{context.produce.acks == -1}}"
            action = "reject"
          }, 

          {
            description = "Transactional ID must start with \"team-a.\""
            rule = "{{dyn(context.produce.transactional_id) == null || context.produce.transactional_id.startsWith(\"team-a.\")}}"
            action = "reject"
          }        ]
      }, 

      {
        type = "fetch"
        rules = [
          {
            description = "Consumers must use the read_committed isolation level"
            rule = "{{context.fetch.isolation_level == \"read_committed\"}}"
            action = "reject"
          }, 

          {
            description = "fetch.max.wait.ms must be at least 100 ms"
            rule = "{{context.fetch.max_wait_ms >= 100}}"
            action = "reject"
          }, 

          {
            description = "Fetch responses can be at most 50 MiB"
            rule = "{{context.fetch.max_bytes <= 52428800}}"
            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: produce_fetch_settings_policy
        type: request_rule_validator
        request_rule_validator:
          name: produce_fetch_settings_policy
          config:
            requests:
            - type: produce
              rules:
              - description: Producers must use acks=all
                rule: "{{context.produce.acks == -1}}"
                action: reject
              - description: Transactional ID must start with "team-a."
                rule: '{{dyn(context.produce.transactional_id) == null || context.produce.transactional_id.startsWith("team-a.")}}'
                action: reject
            - type: fetch
              rules:
              - description: Consumers must use the read_committed isolation level
                rule: '{{context.fetch.isolation_level == "read_committed"}}'
                action: reject
              - description: fetch.max.wait.ms must be at least 100 ms
                rule: "{{context.fetch.max_wait_ms >= 100}}"
                action: reject
              - description: Fetch responses can be at most 50 MiB
                rule: "{{context.fetch.max_bytes <= 52428800}}"
                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 producer and consumer settingsv1.3+

Set up the policy

Help us make these docs great!

Still need help?