Record Transcode Produce: Accept JSON and Avro records and store them as Avro - Policy

Accept JSON and Avro records and store them as Avrov1.3+

Use a schema registry with schema_type any so clients can produce either JSON or Avro, and store only Avro in the backend cluster.

The parent Schema Validation policy validates the record value against a schema registry whose schema_type is any, so a client can produce records in either JSON or Avro.

With output_format: avro, this policy converts JSON records to Avro and leaves records that are already Avro unchanged. It looks up the target Avro schema in the same registry by subject and version, and prefixes converted records with the schema reference in the Confluent wire format, so any Confluent-compatible consumer can deserialize them.

This is useful while producers migrate from JSON to Avro and both formats are in flight for a period of time.

Prerequisites

A Schema Validation Produce policy that validates the record value against a schema registry configured with schema_type: any.

Environment variables

PARENT_POLICY_ID: The UUID of the parent Schema Validation policy.

Set up the policy

curl -X POST https://{region}.api.konghq.com/v1/event-gateways/{eventGatewayId}/virtual-clusters/{virtualClusterId}/produce-policies \
    --header "accept: application/json" \
    --header "Content-Type: application/json" \
    --header "Authorization: Bearer $KONNECT_TOKEN" \
    --data '
    {
      "name": "accept-produce-json-and-avro",
      "type": "transcode",
      "parent_policy_id": "'$PARENT_POLICY_ID'",
      "config": {
        "failure_mode": "reject",
        "output_format": "avro",
        "schema_source": {
          "type": "reference",
          "schema_registry": {
            "name": "my-schema-registry"
          },
          "subject": "${context.topic.name + '-value'}",
          "version": "${'latest'}",
          "schema_ref_destination": {
            "type": "confluent_format"
          }
        }
      }
    }
    '

Copied!

Make sure to replace the following placeholders with your own values:

region: Geographic region where your Kong Konnect is hosted and operates.
KONNECT_TOKEN: Your Personal Access Token (PAT) associated with your Konnect account.
virtualClusterId: The id of the Virtual Cluster.
eventGatewayId: The id of the Event Gateway.
eventGatewayListenerId: The id of the Event Gateway Listener.

See the Konnect Event Gateway API reference to learn about region-specific URLs and personal access tokens.

Prerequisite: Configure your Personal Access Token

terraform {
  required_providers {
    konnect-beta = {
      source  = "kong/konnect-beta"
    }
  }
}

provider "konnect-beta" {
  personal_access_token = "$KONNECT_TOKEN"
  server_url            = "https://us.api.konghq.com/"
}

Copied!

Add the following to your Terraform configuration:

resource "konnect_event_gateway_produce_policy_transcode" "my_virtual_cluster_policy_transcode" {
  provider = konnect-beta
  type = "transcode"
  parent_policy_id = var.parent_policy_id
  config = {
    failure_mode = "reject"
    output_format = "avro"

    schema_source = {
      type = "reference"

      schema_registry = {
        name = "my-schema-registry"
      }
      subject = "${context.topic.name + '-value'}"
      version = "${'latest'}"

      schema_ref_destination = {
        type = "confluent_format"
      }
    }
  }


  virtual_cluster_id = konnect_event_gateway_virtual_cluster.my_virtual_cluster.id

  gateway_id = konnect_event_gateway.my_event_gateway.id
}

Copied!

This example requires the following variables to be added to your manifest. You can specify values at runtime by setting TF_VAR_name=value.

variable "parent_policy_id" {
  type = string
}

Copied!

The following example creates a new transcode policy.

kongctl

event_gateway_policy.yaml

Copied!

event_gateways:
  - ref: eventGatewayName
    name: eventGatewayName
    virtual_clusters:
    - ref: virtualClusterName
      name: virtualClusterName
      produce_policies:
      - ref: accept-produce-json-and-avro
        type: transcode
        transcode:
          name: accept-produce-json-and-avro
          parent_policy_id: !env PARENT_POLICY_ID
          config:
            failure_mode: reject
            output_format: avro
            schema_source:
              type: reference
              schema_registry:
                name: my-schema-registry
              subject: "${context.topic.name + '-value'}"
              version: "${'latest'}"
              schema_ref_destination:
                type: confluent_format

Make sure to replace the following placeholders with your own values:

eventGatewayName: The name of your Event Gateway.
virtualClusterName: The name of the Virtual Cluster.

Record Transcode Produce

Accept JSON and Avro records and store them as Avrov1.3+

Prerequisites

Environment variables

Set up the policy

Help us make these docs great!

Still need help?