Home / Event Gateway
Edit
Report

Mask Kafka message fields selected by schema registry tags with Kong Event Gateway

Uses: Event Gateway Konnect API
Incompatible with
on-prem
Related Documentation
Event Gateway Documentation
Tags
#event-gateway #kafka #confluent #schema-registry #avro
Related Resources
Mask sensitive fields in Kafka messages
Schema Validation policy
Schema Registry entity
Validate Avro messages with Confluent Schema Registry
Encrypt Kafka message fields selected by schema registry tags
Minimum Version
Event Gateway - 1.3
TL;DR
  1. Register an Avro schema in a Confluent Schema Registry and tag the sensitive fields.
  2. Create a Schema Validation policy (consume phase) to parse records against the registry.
  3. Nest a Mask Fields policy that computes the fields to mask from the schema tags.

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 redact fields of Kafka messages based on tags stored in a Confluent Schema Registry.

Instead of listing field paths in the policy, you tag the sensitive fields in the schema itself. The Event Gateway reads the tags of each record’s schema and masks every tagged field. When the schema evolves and a new field gets tagged, the Event Gateway masks it without any policy change.

We’ll use a customers topic that holds customer records with personal data. Producers write complete records to Kafka. Consumers that connect through the virtual cluster receive the same records with the tagged fields redacted, so the raw data stays intact in the broker.

Here’s how the data flows through the system:

 
flowchart LR
    P[Producer] --> K[Kafka 
Broker
raw records]
    SR[(Schema Registry
fields tagged PII)] -.-> SV

    subgraph consume [Event Gateway Consume policy chain]
        SV[Schema Validation
Confluent Schema Registry] --> MF[Mask Fields
paths from PII tags]
    end

    K --> SV
    MF --> CO[Consumer
masked records]

The Mask Fields policy only works on string fields. A field that doesn’t exist is ignored. A field that isn’t a string is a policy failure, handled by the failure_mode setting.

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 consumers 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": "customers_vc",
       "destination": {
         "id": "'$BACKEND_CLUSTER_ID'"
       },
       "dns_label": "customers",
       "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": "customers_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_customers_vc",
       "config": {
         "type": "port_mapping",
         "advertised_host": "localhost",
         "destination": {
           "id": "'$VIRTUAL_CLUSTER_ID'"
         }
       }
     }'

Create a Schema Registry entity

Create a Schema Registry entity that points to the Confluent Schema Registry running locally. Since the Event Gateway data plane runs in the same Docker network as the Schema Registry, use the container hostname schema-registry:

SCHEMA_REGISTRY_ID=$(curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/schema-registries" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "name": "local-schema-registry",
       "type": "confluent",
       "config": {
         "schema_type": "avro",
         "endpoint": "http://schema-registry:8081",
         "timeout_seconds": 10
       }
     }' | jq -r ".id"
)

Register an Avro schema with PII tags

Register an Avro schema for the customers topic in the Confluent Schema Registry. The schema defines four fields: name, email, ssn, and city.

Alongside the schema, the registration body carries a metadata block that tags the three personal fields as PII:

cat <<EOF > schema.json
{
  "schemaType": "AVRO",
  "schema": "{\"type\": \"record\", \"name\": \"Customer\", \"fields\": [{\"name\": \"name\", \"type\": \"string\"}, {\"name\": \"email\", \"type\": \"string\"}, {\"name\": \"ssn\", \"type\": \"string\"}, {\"name\": \"city\", \"type\": \"string\"}]}",
  "metadata": {
    "properties": {"owner": "compliance-team"},
    "tags": {
      "name": ["PII"],
      "email": ["PII"],
      "ssn": ["PII"]
    }
  }
}
EOF
curl -sS --fail -X POST http://localhost:8081/subjects/customers-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d @schema.json

The subject name customers-value follows Confluent’s default TopicNameStrategy, which uses the pattern <topic>-value.

Each key of tags is a field path in the record, and each value lists the tags carried by that field. Use the same path that the field has in the message, for example profile.ssn for a field nested under profile. The Event Gateway passes the tag keys through as-is, so a key that doesn’t match a field path in the record is never masked.

Create a Schema Validation policy

Create a Schema Validation policy that parses Avro records against the Confluent Schema Registry during the consume phase. The Mask Fields policy needs a parsed record and the schema tags, so it must be nested under this policy:

SCHEMA_VALIDATION_POLICY_ID=$(curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/virtual-clusters/$VIRTUAL_CLUSTER_ID/consume-policies" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "type": "schema_validation",
       "name": "validate_avro",
       "config": {
         "type": "confluent_schema_registry",
         "schema_registry": {
           "name": "local-schema-registry"
         },
         "validate_value": true,
         "failure_mode": "skip"
       }
     }' | jq -r ".id"
)

The validate_value: true setting validates the record value against the schema, and failure_mode: skip means a record that can’t be decoded or doesn’t conform to the schema is never delivered to the consumer. This fails closed: a record the Event Gateway can’t parse also can’t be masked, so it must not reach the client.

Create a Mask Fields policy

Create the Mask Fields policy nested under the Schema Validation policy. Instead of static paths, the paths field holds an expression that reads the PII tag from the record’s schema metadata and returns the tagged field paths:

curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/virtual-clusters/$VIRTUAL_CLUSTER_ID/consume-policies" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "type": "mask_fields",
       "name": "mask_pii_from_schema",
       "parent_policy_id": "'$SCHEMA_VALIDATION_POLICY_ID'",
       "config": {
         "failure_mode": "skip",
         "mask_fields": [
           {
             "paths": "record.value.schema.metadata.tags[\"PII\"]",
             "strategy": {
               "type": "replace",
               "phrase": "REDACTED"
             }
           }
         ]
       }
     }'

record.value.schema.metadata.tags is a map from tag name to the field paths carrying that tag. It’s populated by the parent Schema Validation policy, so the expression returns ["name", "email", "ssn"] for records that use the registered schema. The strategy applies to all returned paths, so every tagged field is replaced with REDACTED. The untagged city field passes through unchanged.

The policy uses failure_mode: skip, so a record the policy can’t mask is never delivered. For a redaction policy, choose between skip and error to not let unmasked data through. We generally recommend using skip, because error blocks the whole batch and leaves consumers stuck on the problematic offset until someone intervenes.

Configure kafkactl

Create a kafkactl configuration with a direct context that connects to Kafka, and a vc context that connects through the virtual cluster. Both contexts point to the Schema Registry, so kafkactl can serialize records to Avro and deserialize them for display:

cat <<EOF > kafkactl.yaml
contexts:
  direct:
    brokers:
      - localhost:9094
      - localhost:9095
      - localhost:9096
    schemaRegistry:
      url: http://localhost:8081
  vc:
    brokers:
      - localhost:19092
    schemaRegistry:
      url: http://localhost:8081
EOF

Create a topic and produce records

Create the customers topic:

kafkactl -C kafkactl.yaml --context direct create topic customers

Produce two customer records through the virtual cluster. kafkactl serializes them to Avro using the schema from the registry:

echo '{"name":"John Doe","email":"john.doe@example.com","ssn":"098-76-5432","city":"San Francisco"}
{"name":"Maria Silva","email":"maria.silva@example.com","ssn":"123-45-6789","city":"Boston"}' | kafkactl -C kafkactl.yaml --context vc produce customers

Validate

Consume directly from Kafka

Consume the records straight from the broker to confirm that Kafka stores them unchanged:

kafkactl -C kafkactl.yaml --context direct consume customers --from-beginning --exit

Both records come back with the personal data in plaintext:

{"name":"John Doe","email":"john.doe@example.com","ssn":"098-76-5432","city":"San Francisco"}
{"name":"Maria Silva","email":"maria.silva@example.com","ssn":"123-45-6789","city":"Boston"}

Consume through the virtual cluster

Consume the same records through the virtual cluster:

kafkactl -C kafkactl.yaml --context vc consume customers --from-beginning --exit

The tagged fields are redacted, and city is untouched:

{"name":"REDACTED","email":"REDACTED","ssn":"REDACTED","city":"San Francisco"}
{"name":"REDACTED","email":"REDACTED","ssn":"REDACTED","city":"Boston"}

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