Produce Kafka records as Avro with Kong Event Gateway

In this guide, you’ll learn how to let producers send records as plain JSON or as Avro while Event Gateway stores them as Avro in the backend cluster.

Not every producer wants to run a Schema Registry client or serialize Avro directly, and during a migration from JSON to Avro, both formats can be in use at once. The Event Gateway Record Transcode Produce policy converts an already schema-validated record to a different format on its way to the backend cluster. The cluster and every consumer of the topic only ever see a single format, regardless of what the producers send.

We’ll use an orders topic that holds order records. One producer connects through the virtual cluster and sends plain JSON, and another sends Avro serialized with a Schema Registry client. Event Gateway accepts both and stores every record as Avro.

Here’s how the data flows through the system:

 
flowchart LR
    P1[Producer
JSON] --> SV
    P2[Producer
Avro] --> SV

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

    RT --> K[Kafka 
Broker
Avro records]
  

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 that the producer 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 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-19105"
       ]
     }' | 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 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": "any",
         "endpoint": "http://schema-registry:8081",
         "timeout_seconds": 10
       }
     }' | jq -r ".id"
)

The schema_type: any setting lets this one registry accept records in more than one format. Values that carry a schema reference, in the Confluent wire format or in a record header, are decoded using the registry. Any other value is parsed as JSON without a schema. Both the Schema Validation policy and the Record Transcode Produce policy reference this same registry.

Register the Avro schema that Event Gateway converts records into, under the orders-value subject 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. The Record Transcode Produce policy looks up this subject at runtime, so the schema must already exist before a producer sends records. The Avro producer serializes against this same schema, so one registration serves both.

Create a Schema Validation policy that parses a producer’s records during the produce phase. The Record Transcode Produce 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/produce-policies" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "type": "schema_validation",
       "name": "validate_records",
       "config": {
         "type": "confluent_schema_registry",
         "schema_registry": {
           "name": "local-schema-registry"
         },
         "value_validation_action": "reject"
       }
     }' | jq -r ".id"
)

Because the registry’s schema_type is any, this policy accepts both formats at once: a record that references a schema in the registry is decoded and validated against it, and a record with no schema reference is parsed as JSON. The value_validation_action: reject setting means a batch that holds a record that can’t be parsed or validated is rejected outright.

Create the Record Transcode Produce policy nested under the Schema Validation policy. Set output_format: avro to convert each parsed JSON record to Avro before Event Gateway writes it to the backend cluster:

curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/virtual-clusters/$VIRTUAL_CLUSTER_ID/produce-policies" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "type": "transcode",
       "name": "convert_orders_to_avro",
       "parent_policy_id": "'$SCHEMA_VALIDATION_POLICY_ID'",
       "config": {
         "failure_mode": "reject",
         "output_format": "avro",
         "schema_source": {
           "type": "reference",
           "schema_registry": {
             "name": "local-schema-registry"
           },
           "subject": "context.topic.name + \"-value\"",
           "version": "\"latest\"",
           "schema_ref_destination": {
             "type": "confluent_format"
           }
         }
       }
     }'

In this configuration:

• schema_source: Required to convert to Avro, because Avro needs a schema to serialize the record value.
• subject: An expression that selects the subject. In this example, context.topic.name + '-value' resolves to orders-value for records produced to the orders topic.
• version: An expression that selects the schema version. In this example, 'latest' resolves to the newest registered version.
• schema_ref_destination: confluent_format: Prefixes the converted record with a reference to the schema, so any Confluent-compatible consumer of the topic can deserialize it.
• failure_mode: reject: Used by both the Schema Validation policy and the nested Record Transcode Produce policy. If either policy fails to process a record, the whole batch is rejected and never written to the backend cluster. The alternatives are passthrough and mark.

Create a kafkactl configuration with three contexts:

• vc: produces through the virtual cluster with no Schema Registry configured, for the producer that sends plain JSON.
• vc-avro: produces through the same virtual cluster with the Schema Registry configured, for the producer that sends Avro.
• direct: connects straight to Kafka using the Schema Registry for Avro deserialization.

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

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

Create the orders topic:

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

Produce two order records as plain JSON through the virtual cluster:

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

A second producer sends the same kind of record as Avro, using a client that serializes values with the Schema Registry. Switch to the vc-avro context, which serializes values as Avro and prefixes them with a schema reference in the Confluent wire format:

kafkactl -C kafkactl.yaml --context vc-avro produce orders \
  --value='{"order_id":"1003","item":"sprocket","amount":7.25}'

The Schema Validation policy accepts this record too, because it references the Order schema registered in the Schema Registry. The record value already matches the Record Transcode Produce policy’s output_format, so the policy leaves it unchanged, and the backend cluster stores the record exactly as the producer sent it.

Both producers now write to the same topic through the same policy chain: one sends plain JSON, the other sends Avro, and both records are accepted.

Consume the records straight from the broker to confirm that Kafka stores every record as Avro, no matter which format the producers sent. The --print-schema flag displays the Avro schema used for deserialization:

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

All three 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}
##{"type":"record","name":"Order","fields":[{"name":"order_id","type":"string"},{"name":"item","type":"string"},{"name":"amount","type":"double"}]}#1#{"order_id":"1003","item":"sprocket","amount":7.25}

The Record Transcode Produce policy converted the records from the JSON producer to Avro, and the record from the Avro producer passed through unchanged. In both cases, the backend cluster and every consumer of the topic only ever see Avro.