Home / Kong Operator / Gateway Deployment / How-To
Edit
Report

Validate Kafka message schemas with EventGatewaySchemaRegistry

Uses: Kong Operator
Incompatible with
on-prem
Related Documentation
Operator Documentation
Related Resources
Deploy Kong Event Gateway with port mapping
Kong Event Gateway with Kong Operator
Policies
Minimum Version
Kong Operator - 2.3
TL;DR

Create an EventGatewaySchemaRegistry resource pointing at your schema registry, then reference it from an EventGatewayVirtualClusterConsumePolicy or EventGatewayVirtualClusterProducePolicy with a schemaValidation config.

Prerequisites

Kubernetes namespaces

Create the namespaces used throughout this tutorial:

kubectl create namespace kong
kubectl create namespace kafka

Kafka cluster

The schema registry stores its schemas in Kafka, and Kong Event Gateway needs a backend cluster to proxy. Deploy a three-broker Kafka cluster with the Bitnami chart:

  1. Add the Bitnami Helm repository:

    helm repo add bitnami https://charts.bitnami.com/bitnami
helm repo update

  2. Write the Kafka configuration file:

    cat <<'EOF' >/tmp/kafka-values.yaml
image:
  registry: docker.io
  repository: bitnamilegacy/kafka
  tag: 4.0.0-debian-12-r6

listeners:
  client:
    protocol: PLAINTEXT
externalAccess:
  enabled: false
kraft:
  enabled: true
controller:
  replicaCount: 3
broker:
  replicaCount: 0
EOF

  3. Install Kafka:

    helm install kafka-cluster bitnami/kafka \
  -n kafka \
  --version 32.4.3 \
  -f /tmp/kafka-values.yaml

  4. Wait for all Kafka brokers to be ready:

    kubectl wait pod -n kafka \
  --for=condition=Ready \
  --selector app.kubernetes.io/name=kafka \
  --timeout=5m

Confluent Schema Registry

Deploy a Confluent Schema Registry backed by the Kafka cluster. Running it in the cluster means Kong Event Gateway can reach it at http://schema-registry.kafka.svc.cluster.local:8081:

  1. Deploy the registry and its Service:

    echo '
apiVersion: apps/v1
kind: Deployment
metadata:
  name: schema-registry
  namespace: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: schema-registry
  template:
    metadata:
      labels:
        app: schema-registry
    spec:
      enableServiceLinks: false
      containers:
        - name: schema-registry
          image: confluentinc/cp-schema-registry:8.2.1
          ports:
            - containerPort: 8081
          env:
            - name: SCHEMA_REGISTRY_HOST_NAME
              value: schema-registry
            - name: SCHEMA_REGISTRY_LISTENERS
              value: http://0.0.0.0:8081
            - name: SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS
              value: kafka-cluster.kafka.svc.cluster.local:9092
---
apiVersion: v1
kind: Service
metadata:
  name: schema-registry
  namespace: kafka
spec:
  selector:
    app: schema-registry
  ports:
    - name: http
      port: 8081
      targetPort: 8081
' | kubectl apply -f -

  2. Wait for the registry to be ready:

    kubectl wait pod -n kafka \
  --for=condition=Ready \
  --selector app=schema-registry \
  --timeout=5m

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 Operator running

  1. Add the Kong Helm charts:

    helm repo add kong https://charts.konghq.com
helm repo update

  2. Install Kong Operator using Helm:

    helm upgrade --install kong-operator kong/kong-operator -n kong-system \
  --create-namespace \
  --set image.tag=2.3 \
  --set env.ENABLE_CONTROLLER_KONNECT=true \
  --set env.ENABLE_CONTROLLER_KEGDATAPLANE=true

    If you want cert-manager to issue and rotate the admission and conversion webhook certificates, install cert-manager to your cluster and enable cert-manager integration by passing the following argument while installing, in the next step:

    --set global.webhooks.options.certManager.enabled=true

    If you do not enable this, the chart will generate and inject self-signed certificates automatically. We recommend enabling cert-manager to manage the lifecycle of these certificates. Kong Operator needs a certificate authority to sign the certificate for mTLS communication between the control plane and the data plane. This is handled automatically by the Helm chart. If you need to provide a custom CA certificate, refer to the certificateAuthority section in the values.yaml of the Helm chart to learn how to create and reference your own CA certificate.

Create a KonnectAPIAuthConfiguration resource

 
kubectl create namespace kong --dry-run=client -o yaml | kubectl apply -f -
echo '
kind: KonnectAPIAuthConfiguration
apiVersion: konnect.konghq.com/v1alpha1
metadata:
  name: konnect-api-auth
  namespace: kong
spec:
  type: token
  token: "'$KONNECT_TOKEN'"
  serverURL: us.api.konghq.com
' | kubectl apply -f -

Kong Event Gateway control plane, backend cluster, and virtual cluster

The schema registry and the schema validation policy attach to a Kong Event Gateway control plane and virtual cluster. Create them, along with the backend cluster that points at Kafka:

  1. Create the KonnectEventGateway resource:

    echo '
apiVersion: konnect.konghq.com/v1alpha1
kind: KonnectEventGateway
metadata:
  name: cp-event-1
  namespace: kong
spec:
  apiSpec:
    name: cp-event-1
    description: Event Gateway control plane managed by Kubernetes
  konnect:
    authRef:
      name: konnect-api-auth
' | kubectl apply -f -

kubectl wait konnecteventgateway/cp-event-1 -n kong \
  --for=condition=Programmed=True \
  --timeout=10m

  2. Create the EventGatewayBackendCluster resource:

    echo '
apiVersion: configuration.konghq.com/v1alpha1
kind: EventGatewayBackendCluster
metadata:
  name: default-backend-cluster
  namespace: kong
spec:
  gatewayRef:
    type: namespacedRef
    namespacedRef:
      name: cp-event-1
  apiSpec:
    name: default_backend_cluster
    bootstrapServers:
      - kafka-cluster.kafka.svc.cluster.local:9092
    authentication:
      type: anonymous
      anonymous: {}
    insecureAllowAnonymousVirtualClusterAuth: Enabled
    tls:
      enabled: Disabled
' | kubectl apply -f -

kubectl wait eventgatewaybackendcluster/default-backend-cluster -n kong \
  --for=condition=Programmed=True \
  --timeout=10m

  3. Create the EventGatewayVirtualCluster resource:

    echo '
apiVersion: configuration.konghq.com/v1alpha1
kind: EventGatewayVirtualCluster
metadata:
  name: example-virtual-cluster
  namespace: kong
spec:
  eventGatewayBackendClusterRef:
    type: namespacedRef
    namespacedRef:
      name: default-backend-cluster
  apiSpec:
    name: example_virtual_cluster
    dnsLabel: vcluster-1
    aclMode: passthrough
    authentication:
      - type: anonymous
    namespace:
      prefix: "vc1_"
      mode: hide_prefix
' | kubectl apply -f -

kubectl wait eventgatewayvirtualcluster/example-virtual-cluster -n kong \
  --for=condition=Programmed=True \
  --timeout=10m

EventGatewaySchemaRegistry connects Kong Event Gateway to a schema registry so consume and produce policies can validate Kafka record schemas before they reach clients or your backend cluster.

The examples below use the cp-event-1 control plane, example-virtual-cluster virtual cluster, and schema-registry registry Service created in the prerequisites.

Create the EventGatewaySchemaRegistry

  1. Create the EventGatewaySchemaRegistry resource, pointing at the schema registry endpoint:

    echo '
apiVersion: configuration.konghq.com/v1alpha1
kind: EventGatewaySchemaRegistry
metadata:
  name: example-schema-registry
  namespace: kong
spec:
  gatewayRef:
    type: namespacedRef
    namespacedRef:
      name: cp-event-1
  apiSpec:
    type: confluent
    confluent:
      name: example_schema_registry
      description: Schema registry for example_virtual_cluster
      config:
        endpoint: http://schema-registry.kafka.svc.cluster.local:8081
        schemaType: json
        timeoutSeconds: 10
' | kubectl apply -f -

  2. Wait for the resource to be ready:

    kubectl wait eventgatewayschemaregistry/example-schema-registry -n kong \
  --for=condition=Programmed=True \
  --timeout=10m

Enforce schema validation on a virtual cluster

Reference the EventGatewaySchemaRegistry from an EventGatewayVirtualClusterConsumePolicy using type: schemaValidation. Kong Event Gateway supports two validation modes, depending on how records are serialized.

This example uses confluentSchemaRegistry, for producers that serialize records with the Confluent wire format (a schema ID embedded in the record header). Kong Event Gateway resolves the schema by ID from the registry:

echo '
apiVersion: configuration.konghq.com/v1alpha1
kind: EventGatewayVirtualClusterConsumePolicy
metadata:
  name: example-schema-validation-confluent
  namespace: kong
spec:
  eventGatewayVirtualClusterRef:
    type: namespacedRef
    namespacedRef:
      name: example-virtual-cluster
  apiSpec:
    type: schemaValidation
    schemaValidation:
      name: example_schema_validation_confluent
      description: Validate Confluent wire-format records against the registry
      config:
        type: confluentSchemaRegistry
        confluentSchemaRegistry:
          schemaRegistry:
            kind: EventGatewaySchemaRegistry
            name: example-schema-registry
          keyValidationAction: mark
          valueValidationAction: mark
' | kubectl apply -f -

If your producers send plain JSON records without a wire-format schema ID, use the json mode instead. For more information, see the FAQs.

A virtual cluster should have a single, consistent schema validation policy, so apply only one EventGatewayVirtualClusterConsumePolicy with type: schemaValidation at a time. The same type: schemaValidation config also works on EventGatewayVirtualClusterProducePolicy to validate records before they reach the backend cluster.

Wait for the resource to be ready:

kubectl wait eventgatewayvirtualclusterconsumepolicy/example-schema-validation-confluent -n kong \
  --for=condition=Programmed=True \
  --timeout=10m

Validate

Check that the policy reconciled without errors:

kubectl describe eventgatewayvirtualclusterconsumepolicy/example-schema-validation-confluent -n kong

A Programmed=True condition confirms Kong Event Gateway accepted the policy and is enforcing schema validation on the example-virtual-cluster virtual cluster. Records that fail validation are either marked with a kong/server header or skipped so they aren’t delivered to the client, depending on whether you set mark or skip in keyValidationAction and valueValidationAction.

FAQs

How do I validate plain JSON records?

Use json instead of confluentSchemaRegistry when producers send plain JSON records without a wire-format schema ID. This mode only checks that each record is valid JSON, so it doesn’t consult a schema registry. Omit schemaRegistry entirely, Konnect rejects it when type is json.

Apply this policy in place of the confluentSchemaRegistry one, not alongside it:

echo '
apiVersion: configuration.konghq.com/v1alpha1
kind: EventGatewayVirtualClusterConsumePolicy
metadata:
  name: example-schema-validation-json
  namespace: kong
spec:
  eventGatewayVirtualClusterRef:
    type: namespacedRef
    namespacedRef:
      name: example-virtual-cluster
  apiSpec:
    type: schemaValidation
    schemaValidation:
      name: example_schema_validation_json
      description: Validate that consumed records are valid JSON
      config:
        type: json
        json:
          valueValidationAction: skip
' | kubectl apply -f -

How do I connect to a schema registry that requires authentication?

The registry deployed in the prerequisites accepts unauthenticated requests, so the authentication field is omitted in this tutorial. If your own registry requires basic authentication, store the password in a Kubernetes Secret. Kong Operator only watches Secrets labeled konghq.com/secret="true":

kubectl create secret generic schema-registry-password \
  --from-literal=password='my-schema-registry-password' \
  -n kong
kubectl label secret schema-registry-password -n kong konghq.com/secret="true"

Then add an authentication block to config that references the Secret:

config:
  endpoint: https://schema-registry.example.com
  schemaType: json
  timeoutSeconds: 10
  authentication:
    type: basic
    basic:
      username: schema-registry-user
      password:
        type: secretRef
        secretRef:
          name: schema-registry-password
          key: password
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