Home / Event Gateway
Edit
Report

Consume Kafka records as JSON 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
Record Transcode Consume policy
Schema Validation policy
Schema Registry entity
Validate Avro messages with Confluent Schema Registry
Minimum Version
Event Gateway - 1.3
TL;DR
  1. Create a Schema Validation policy (consume phase) to parse Avro records against a Confluent Schema Registry.
  2. Nest a Record Transcode Consume policy that converts the parsed record to JSON.

Prerequisites

Install kafkactl

Install kafkactl. You’ll need it to interact with Kafka clusters.

Start a local Kafka cluster and schema registry

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 store Kafka records as Avro and serve them as JSON to a consumer that doesn’t run a Schema Registry client.

Avro keeps records compact and schema-governed in the cluster, but not every consumer wants the overhead of an Avro deserializer. The Event Gateway Record Transcode Consume policy converts an already schema-validated record to a different format on its way to the client, so producers and the cluster keep using Avro while this consumer reads plain JSON.

We’ll use an orders topic that holds order records, produced and stored as Avro. A consumer that connects through the virtual cluster receives the same records converted to JSON, with no producer or cluster changes.

Here’s how the data flows through the system:

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

    subgraph consume [Event Gateway Consume policy chain]
        SV[Schema 
Validation
Parse Avro] --> RT[Record Transcode
convert to JSON]
    end

    K --> SV
    RT --> CO[Consumer
JSON records]

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 the consumer connects 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": "orders_vc",
       "destination": {
         "id": "'$BACKEND_CLUSTER_ID'"
       },
       "dns_label": "orders",
       "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": "orders_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_orders_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 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

Register an Avro schema for the orders topic in the Confluent Schema Registry. The schema defines three fields: order_id, item, and amount:

curl -sS --fail -X POST http://localhost:8081/subjects/orders-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "{\"type\": \"record\", \"name\": \"Order\", \"fields\": [{\"name\": \"order_id\", \"type\": \"string\"}, {\"name\": \"item\", \"type\": \"string\"}, {\"name\": \"amount\", \"type\": \"double\"}]}"}'

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

Create a Schema Validation policy

Create a Schema Validation policy that parses Avro records against the Confluent Schema Registry during the consume phase. The Record Transcode Consume 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": "validate_avro",
       "config": {
         "type": "confluent_schema_registry",
         "schema_registry": {
           "name": "local-schema-registry"
         },
         "value_validation_action": "skip"
       }
     }' | jq -r ".id"
)

The value_validation_action: skip setting means a record that can’t be decoded against the registered Avro schema is never delivered to the consumer.

Create a Record Transcode Consume policy

Create the Record Transcode Consume policy nested under the Schema Validation policy. Set output_format: json to convert each parsed Avro record to JSON before it reaches the consumer:

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": "transcode",
       "name": "convert_orders_to_json",
       "parent_policy_id": "'$SCHEMA_VALIDATION_POLICY_ID'",
       "config": {
         "failure_mode": "skip",
         "output_format": "json"
       }
     }'

Converting to JSON doesn’t require a schema_source, because JSON doesn’t need a schema to serialize the record value.

Both policies use failure_mode: skip, so a record either policy can’t process is never delivered. The alternatives are error, passthrough, and mark.

We generally recommend using skip over error, because error blocks the whole batch and leaves the consumer stuck on the problematic offset until someone intervenes.

Configure kafkactl

Create a kafkactl configuration with a direct context that connects to Kafka using the Schema Registry for Avro serialization, and a vc context that connects through the virtual cluster:

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

The vc context has no Schema Registry configured, because the consumer reads plain JSON through the virtual cluster and doesn’t need one.

Create a topic and produce records

Create the orders topic:

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

Produce two order records directly to Kafka. kafkactl serializes them to Avro using the schema from the registry:

echo '{"order_id":"1001","item":"widget","amount":42.5}
{"order_id":"1002","item":"gadget","amount":19.99}' | kafkactl -C kafkactl.yaml --context direct produce orders

Validate

Consume directly from Kafka

Consume the records straight from the broker to confirm that Kafka stores them as Avro. The --print-schema flag displays the Avro schema used for deserialization:

kafkactl -C kafkactl.yaml --context direct consume orders --from-beginning --exit --print-schema

Both records come back deserialized from Avro, alongside the schema used to decode them:

##{"type":"record","name":"Order","fields":[{"name":"order_id","type":"string"},{"name":"item","type":"string"},{"name":"amount","type":"double"}]}#1#{"order_id":"1001","item":"widget","amount":42.5}
##{"type":"record","name":"Order","fields":[{"name":"order_id","type":"string"},{"name":"item","type":"string"},{"name":"amount","type":"double"}]}#1#{"order_id":"1002","item":"gadget","amount":19.99}

Consume through the virtual cluster

Now consume the same records through the virtual cluster, with no Schema Registry configured for this context:

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

The records arrive as plain JSON, with no Avro schema needed to read them:

{"order_id":"1001","item":"widget","amount":42.5}
{"order_id":"1002","item":"gadget","amount":19.99}

In this case, the producer and the backend cluster never changed formats. The Schema Validation policy parsed the Avro records, and the Record Transcode Consume policy converted them to JSON for this consumer only.

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