Policies

Policies control how Kafka protocol traffic is modified between the client and the backend cluster.

There are two main types of policies:

• Virtual cluster policies: Transformation and validation policies applied to Kafka messages. Virtual cluster policies break down further into cluster, consume, and produce policies.
• Listener policies: Routing policies that pass traffic to the virtual cluster.

Policies execute in chains. The order in which Kong Event Gateway applies policies to modify messages depends on the policy type, and whether the message is a request or response.

flowchart LR
    
    A[Kafka 
    client]
    B[Listener 
    &#40TCP socket&#41
    + listener policies]
    C@{ shape: processes, label: "Virtual clusters
    + consume, produce, 
    and cluster policies"}
    D[Backend 
    cluster]
    E[Kafka 
    cluster]

    A --> B
    subgraph id1 [Event Gateway]
    B --> C 
    C --> D
    end
    D --> E

    style B stroke:#86e2cc
    style C stroke:#86e2cc

    style id1 rx:7,ry:7

Depending on the type of configuration update, the connection between the Kafka client and the backend can be affected:

• Updates to virtual cluster policies don’t cause a connection drop. Policies reload dynamically and take effect on the next request.
• Updates to listener policies cause a connection drop. When the data plane receives configuration updates to a listener policy, it restarts the running proxy services.

If a connection drop occurs, the Kafka client is designed to handle short-lived breaks in connections.

Virtual cluster policies are applied to Kafka traffic via virtual clusters. They let you modify headers, encrypt or decrypt records, validate schemas, and more.

See the Kong Event Gateway policy hub for all available virtual cluster policies.

Phases

Virtual cluster policies run during specific phases, which represent stages in a record’s lifecycle: cluster, consume, and produce.

Record serialization

Some policies operate on parsed records, while others work with raw serialized data.

• Parsed records: Deserialized into a structured format (for example, JSON objects or Avro records).
• Non-parsed records: Raw, serialized byte data.

Records are deserialized on produce and re-serialized on consume.

Policy nesting

Certain policies can serve as parent policies. You can nest policies within a parent policy to process the record content.

Nested policies are sorted into an index. You can re-order nested policies within the parent policy by adjusting this index.

curl -X POST "https://us.api.konghq.com/v1/event-gateways/{gatewayId}/virtual-clusters/{virtualClusterId}/produce-policies/{policyID}/move" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "index": 2
     }'

Set up a virtual cluster policy

Kong Event Gateway has a few built-in virtual cluster policies, all of which have their own specific configurations and examples. See all Event Gateway policies for their individual configurations.

Here’s an example configuration for the Modify Headers consume policy:

curl -X POST https://{region}.api.konghq.com/v1/event-gateways/{eventGatewayId}/virtual-clusters/{virtualClusterId}/consume-policies \
    --header "accept: application/json" \
    --header "Content-Type: application/json" \
    --header "Authorization: Bearer $KONNECT_TOKEN" \
    --data '
    {
      "name": "new-header",
      "type": "modify-headers",
      "config": {
        "actions": [
          {
            "op": "set",
            "key": "My-New-Header",
            "value": "header_value"
          }
        ]
      }
    }
    '

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

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

resource "konnect_event_gateway_consume_policy_modify_headers" "my_virtual_cluster_policy_modify_headers" {
  provider = konnect-beta
  type = "modify-headers"
  config = {
    actions = [
      {
        op = "set"
        key = "My-New-Header"
        value = "header_value"
      }    ]
  }


  virtual_cluster_id = konnect_event_gateway_virtual_cluster.my_virtual_cluster.id

  gateway_id = konnect_event_gateway.my_event_gateway.id
}

event_gateways:
  - ref: eventGatewayName
    name: eventGatewayName
    virtual_clusters:
    - ref: virtualClusterName
      name: virtualClusterName
      consume_policies:
      - ref: new-header
        type: modify-headers
        modify-headers:
          name: new-header
          config:
            actions:
            - op: set
              key: My-New-Header
              value: header_value

Some produce and consume policies have a failure_mode setting that controls what happens to a record when the policy fails to process it:

See each policy for the failure modes it supports.

mark

The mark failure mode attaches the policy failure reason to the record in a kong/policy-failure-<id> header.

On produce, Event Gateway stores the header with the record in the backend cluster. The header remains on the record, so every consumer of the topic sees it, including systems that never pass through Event Gateway.

A producer with access to the topic can trigger policy failures on purpose and read the failure reasons to learn about your policy configuration.

Warning: Failure reasons can contain runtime-specific information, such as schema IDs, registry URLs, field paths, KMS ARNs and key IDs, KMS error text, JWT issuer and audience data, and CEL expression snippets. If you use mark, configure a Modify Headers policy that runs after the failing policy to remove the kong/policy-failure-<id> header before the record is stored in the backend cluster or delivered to a client.

error

The error failure mode stops the delivery of the whole batch that contains the failing record. The client receives an error with the policy failure reason.

Warning: A single record that always fails the policy, for example one written by a producer with access to the topic, blocks every consumer group at that offset, and recovery requires manual operator action. Use error sparingly, and prefer skip when producers you don’t trust can write to the topic. skip stops only the failing record, so consumer groups keep making progress.

Listener policies are applied to layer 4 TCP traffic on listeners, for example to enforce TLS, select a certificate for the TLS connection, or to route to a specific virtual cluster.

See the Kong Event Gateway policy hub for all available listener policies.

Set up a listener policy

Kong Event Gateway has a few built-in listener policies, all of which have their own specific configurations and examples. See all Event Gateway policies for their individual configurations.

Here’s an example configuration for the Forward to Virtual Cluster policy:

curl -X POST https://{region}.api.konghq.com/v1/event-gateways/{eventGatewayId}/listeners/{eventGatewayListenerId}/policies \
    --header "accept: application/json" \
    --header "Content-Type: application/json" \
    --header "Authorization: Bearer $KONNECT_TOKEN" \
    --data '
    {
      "name": "forward",
      "type": "forward-to-virtual-cluster",
      "config": {
        "advertised_host": "0.0.0.0",
        "bootstrap_port": "at_start",
        "destination": {
          "name": "example-virtual-cluster"
        },
        "min_broker_id": 1,
        "type": "port_mapping"
      }
    }
    '

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

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

resource "konnect_event_gateway_listener_policy_forward_to_virtual_cluster" "my_listener_policy_forward_to_virtual_cluster" {
  provider = konnect-beta
  type = "forward-to-virtual-cluster"
  config = {
    advertised_host = "0.0.0.0"
    bootstrap_port = "at_start"

    destination = {
      name = "example-virtual-cluster"
    }
    min_broker_id = 1
    type = "port_mapping"
  }


  event_gateway_listener_id = konnect_event_gateway_listener.my_listener.id

  gateway_id = konnect_event_gateway.my_event_gateway.id
}

event_gateways:
  - ref: eventGatewayName
    name: eventGatewayName
    listeners:
    - ref: listenerName
      name: listenerName
      policies:
      - ref: forward
        type: forward-to-virtual-cluster
        forward-to-virtual-cluster:
          name: forward
          config:
            advertised_host: 0.0.0.0
            bootstrap_port: at_start
            destination:
              name: example-virtual-cluster
            min_broker_id: 1
            type: port_mapping

Policies have a condition field that determines whether the policy executes or not. By writing conditions using expressions, you can access dynamic configuration from the execution context.

For example, you can create a condition that selects all topics that end with the suffix my_suffix:

"condition": "{{context.topic.name.endsWith('my_suffix')}}"

See the expressions reference for more information.