Home / Event Gateway
Edit
Report

Mask sensitive fields in Kafka messages with Kong Event Gateway

Uses: Event Gateway Konnect API
Incompatible with
on-prem
Related Documentation
Event Gateway Documentation
Tags
#event-gateway #kafka
Related Resources
Mask Fields Consume policy
Schema Validation policy
Virtual clusters
Encrypt and decrypt Kafka message fields
Event Gateway Control Plane API
Minimum Version
Event Gateway - 1.3
TL;DR
  1. Create a Schema Validation policy (consume phase) to parse JSON records.
  2. Nest a Mask Fields policy that selects the sensitive fields and a masking strategy for each one.

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 sensitive fields of Kafka messages before they reach consumers.

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 personal fields redacted, so the raw data stays intact in the broker and remains available to systems that are allowed to see it.

Masking runs in the consume phase, which means the stored records are never modified. If you need data to reach the broker in a redacted state, apply the same policy in the produce phase instead. That change is irreversible: the original values are permanently discarded, since only the redacted version ever reaches Kafka.

The Mask Fields policy supports three strategies, and this guide uses all of them:

Strategy

Description
keep_chars Keeps a number of leading and trailing characters and replaces the middle with a fixed phrase.
email Applies a separate strategy to the local part and to the domain of an email address.
replace Replaces the whole value with a fixed phrase.

Here’s how the data flows through the system:

 
flowchart LR
    P[Producer] --> K[Kafka 
Broker
raw records]

    subgraph consume [Event Gateway Consume policy chain]
        SV[Schema 
Validation
Parse JSON] --> MF[Mask Fields
redact name,
email, ssn]
    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'"
         }
       }
     }'

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 a Schema Validation policy

Create a Schema Validation policy that parses JSON records during the consume phase. The Mask Fields policy needs a parsed record, 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": "parse_json",
       "config": {
         "type": "json",
         "validate_value": true,
         "failure_mode": "skip"
       }
     }' | jq -r ".id"
)

The failure_mode: skip setting means a record whose value isn’t valid JSON 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. Each entry under mask_fields selects a set of field paths and the strategy that redacts them:

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_customer_pii",
       "parent_policy_id": "'$SCHEMA_VALIDATION_POLICY_ID'",
       "config": {
         "failure_mode": "skip",
         "mask_fields": [
           {
             "paths": [
               {
                 "match": "customer.ssn"
               }
             ],
             "strategy": {
               "type": "keep_chars",
               "first": 0,
               "last": 4,
               "phrase": "***-**-"
             }
           },
           {
             "paths": [
               {
                 "match": "customer.email"
               }
             ],
             "strategy": {
               "type": "email",
               "local_part": {
                 "type": "keep_chars",
                 "first": 1,
                 "last": 0,
                 "phrase": "***"
               },
               "domain": {
                 "type": "keep_all"
               }
             }
           },
           {
             "paths": [
               {
                 "match": "customer.name"
               }
             ],
             "strategy": {
               "type": "replace",
               "phrase": "REDACTED"
             }
           }
         ]
       }
     }'

In this configuration:

  • customer.ssn keeps its last four characters, and the rest becomes ***-**-. The phrase has a fixed length, so the masked value doesn’t reveal how long the original was.
  • customer.email keeps the first character of the local part and the whole domain, which is enough to recognize an account without exposing the address.
  • customer.name is replaced entirely, because no part of a name is safe to show.
  • customer.city isn’t selected, so it passes through unchanged.

Both policies use failure_mode: skip, so a record the policy can’t mask is never delivered. The alternatives are error, passthrough, and mark. 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.

Don’t use passthrough or mark here, because both deliver the record with the sensitive fields still readable.

Configure kafkactl

Create a kafkactl configuration with a direct context that connects to Kafka, and a vc context that connects through the virtual cluster:

cat <<EOF > kafkactl.yaml
contexts:
  direct:
    brokers:
      - localhost:9094
      - localhost:9095
      - localhost:9096
  vc:
    brokers:
      - localhost:19092
EOF

Create a topic and produce records

Create the customers topic:

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

Produce three customer records directly to Kafka, bypassing the Event Gateway:

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

The third record holds an ssn that’s a number instead of a string. We’ll use it to show what happens when the policy can’t mask a field.

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

All three records come back with the personal data in plaintext:

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

Consume through the virtual cluster

Now consume the same records through the virtual cluster:

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

The personal fields are redacted, and city is untouched:

{"customer":{"name":"REDACTED","email":"j***@example.com","ssn":"***-**-5432","city":"San Francisco"}}
{"customer":{"name":"REDACTED","email":"m***@example.com","ssn":"***-**-6789","city":"Boston"}}

Only two records arrive. The third one, whose ssn isn’t a string, couldn’t be masked, so failure_mode: skip dropped it instead of delivering it with the personal data exposed.

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