Enforce Centralized Governance with Confluent Cloud Gateway

To enforce centralized governance on your Kafka traffic, you must configure Confluent Cloud Gateway (Confluent Gateway) to communicate with a Confluent Cloud Schema Registry and set the enforcement levels in the gateway.yaml file.

For an overview of how centralized governance enforcement works, see Enforce Data Governance Centrally with Confluent Gateway.

Before you begin

Ensure your environment meets the following requirements:

Gateway Version: Deploy Confluent Gateway version 1.3.
Confluent Cloud Schema Registry: Enable Confluent Cloud Governance with Stream Governance Advanced.
Serializer: Use Confluent Cloud Schema Registry serializers that produce the standard Confluent wire format (a magic byte 0x00 followed by a 4-byte schema ID). For more information, see Serializer and Deserializer Development.
Gateway Authentication: Provision Confluent Gateway with valid Confluent Cloud Schema Registry credentials using Basic Authentication.
Access Control: Grant Confluent Gateway read access to Confluent Cloud Schema Registry so it can read the schemas for all topics that flow through it.

Step 1: Configure the Confluent Cloud Schema Registry connection

Connect Confluent Gateway to Confluent Cloud Schema Registry to enable schema retrieval and validate incoming records.

gateway:
  schemaValidation:
    schemaRegistryUrls:                                     --- [1]
      - "https://psrc-xxxxx.us-east-2.aws.confluent.cloud"
    schemaRegistryConfigs:                                  --- [2]
      basic.auth.credentials.source: "USER_INFO"            --- [3]
      basic.auth.user.info: "<SR-API-KEY>:<SR-API-SECRET>"  --- [4]
    context: null                                           --- [5]
    valueValidationLevel: NONE                              --- [6]
    keyValidationLevel: NONE                                --- [7]
    valueSubjectNameStrategy: TOPIC                         --- [8]
    keySubjectNameStrategy: TOPIC                           --- [9]
    topics: []                                              --- [10]

[1] schemaRegistryUrls: A list of one or more Confluent Cloud Schema Registry URLs. A single Confluent Cloud Schema Registry cluster typically has one URL. If you list more than one, Confluent Gateway connects to the next URL if the primary URL is unreachable.
[2] schemaRegistryConfigs: A container for all configurations passed to the Confluent Cloud Schema Registry client, including security protocols, timeouts, and authentication details such as basic authentication, OAuth, and OpenID Connect (OIDC).
[3] basic.auth.credentials.source: The source of the credentials for basic authentication. Setting this to USER_INFO instructs Confluent Gateway to read the username and password from the basic.auth.user.info property.
[4] basic.auth.user.info: The Confluent Cloud Schema Registry API key and secret, in <SR-API-KEY>:<SR-API-SECRET> format. For production environments, use environment variables or a secrets provider rather than plain text.
[5] context: The Confluent Cloud Schema Registry context to use for schema lookups. Use a context to isolate schemas by environment or tenant. When set to null (default), Confluent Gateway uses the default Confluent Cloud Schema Registry context.
[6] valueValidationLevel: The enforcement level applied to record values on the produce path. Defaults to NONE. For more information, see Step 2: Configure the enforcement level.
[7] keyValidationLevel: The enforcement level applied to record keys. Defaults to NONE. Set this property only if your producers serialize keys with a schema registered in Confluent Cloud Schema Registry.
[8] valueSubjectNameStrategy: The subject name strategy that Confluent Gateway uses to derive the subject name for record values in Confluent Cloud Schema Registry. For more information, see Step 3: Configure the subject name strategy.
[9] keySubjectNameStrategy: The subject name strategy that Confluent Gateway uses to derive the subject name for record keys in Confluent Cloud Schema Registry. Set this property only if your producers serialize keys with a schema registered in Confluent Cloud Schema Registry.
[10] topics: (Optional) A list of per-topic overrides. Each entry can override validation levels, subject name strategies, and the Confluent Cloud Schema Registry context for a specific Kafka topic. For more information, see Step 4: Configure enforcement levels per topic.

Step 2: Configure the enforcement level

Confluent Gateway supports four enforcement levels. Each level adds more validation and causes additional latency. Choose the level that fits your balance between throughput and data integrity.

Enforcement level	Configuration value	Description	When to use
No enforcement	`NONE`	Pass-through mode. Confluent Gateway does not validate produce or fetch requests.	Use during initial rollout, or for topics where another system enforces schema validation.
Schema ID enforcement	`ID`	Producer: Validates that the schema ID embedded in the record exists in Confluent Cloud Schema Registry under the correct subject. Confluent Gateway does not deserialize the record, so the throughput impact is minimal. Consumer: Pass-through, no validation applied.	Use as the default production setting. Confluent Gateway rejects records with an unknown or mismatched schema ID before they reach the broker.
Deep schema validation	`SCHEMA`	Producer: Deserializes the payload and validates every field against the registered schema, then forwards the original bytes to the broker unchanged. Does not apply data contract rules. Consumer: Pass-through, no validation applied.	Use for critical ingress pipelines where malformed payloads could cause downstream failures and `ID` validation alone is not sufficient.
Data contract rules	`SCHEMA_RULES`	Producer: Performs end-to-end field-level validation and applies data contract rules, including field-level encryption. For more details, see Step 5: Configure data contract rules and encryption. Consumer: Confluent Gateway applies decryption rules before delivering records to consumers.	Use when you need field-level encryption or data quality rule enforcement in addition to schema validation.

Add valueValidationLevel to the schemaValidation parameters configured in Step 1: Configure the |sr-ccloud| connection.

gateway:
  schemaValidation:
    valueValidationLevel: <validation level>

valueValidationLevel sets the enforcement level applied to record values on the produce path. Confluent Gateway evaluates every incoming record against this level before forwarding it to the broker. Defaults to NONE.

Step 3: Configure the subject name strategy

valueSubjectNameStrategy controls how Confluent Gateway derives the subject name used to look up a schema in Confluent Cloud Schema Registry. By default, Confluent Gateway constructs the name from the topic name. Update the strategy if your subject naming convention differs from the default or if you want to use a different method for associating records with schemas.

Strategy	How Confluent Gateway constructs the subject name	Example subject name
`TOPIC` (default)	Uses `<topic-name>-value` (or `<topic-name>-key` for keys). Use this when each topic has one schema.	For topic `orders` → `orders-value`
`RECORD`	Uses the fully qualified record name from the schema. Use this when multiple record types share the same topic.	For record `com.example.OrderEvent` → `com.example.OrderEvent`
`TOPIC_RECORD`	Uses `<topic-name>-<record-name>`. Use this when multiple record types share the same topic and you want subject isolation per topic.	For topic `orders` and record `OrderEvent` → `orders-OrderEvent`
`ASSOCIATED`	Looks up the subject name through the Confluent Cloud Schema Registry associations API, and falls back to `TOPIC` when no association exists. Use this strategy when you manage schema-to-topic mappings centrally in Confluent Cloud Schema Registry.	The subject that Confluent Cloud Schema Registry associates with the schema.

Step 4: Configure enforcement levels per topic

Add per-topic overrides to enforce a different validation level on specific topics. By default, the valueValidationLevel you set in Step 2: Configure the enforcement level applies to all topics.

In this example, the global enforcement level is ID, but Confluent Gateway validates the critical-orders topic at the stricter SCHEMA level, and public-events bypasses validation entirely.

gateway:
  schemaValidation:
    schemaRegistryUrls:
      - "https://psrc-xxxxx.us-east-2.aws.confluent.cloud"
    schemaRegistryConfigs:
      basic.auth.credentials.source: "USER_INFO"
      basic.auth.user.info: "<SR-API-KEY>:<SR-API-SECRET>"
    keyValidationLevel: NONE
    valueValidationLevel: ID          --- [1]
    valueSubjectNameStrategy: TOPIC
    topics:
      - name: "critical-orders"       --- [2]
        valueValidationLevel: SCHEMA  --- [3]
      - name: "public-events"         --- [2]
        valueValidationLevel: NONE    --- [3]

[1] The global enforcement level applied to all topics not listed under topics.
[2] topics[].name: The name of the Kafka topic to override.
[3] topics[].valueValidationLevel: The enforcement level for this specific topic. Overrides the global valueValidationLevel. Accepts the same values as discussed in Step 2: Configure the enforcement level: NONE, ID, SCHEMA, SCHEMA_RULES.

Note

Per-topic overrides take precedence over the global settings. Setting a topic’s valueValidationLevel to NONE disables validation for that topic even when you set the global level to ID, SCHEMA, or SCHEMA_RULES.

Step 5: Configure data contract rules and encryption

When you set the enforcement level in valueValidationLevel to SCHEMA_RULES, Confluent Gateway applies data contract rules defined in Confluent Cloud Schema Registry to the incoming records. These rules are commonly used for field-level or full-payload encryption. Confluent Gateway encrypts sensitive data on the produce path before it reaches the broker, and decrypts it on the fetch path before it reaches the consumer.

Prerequisites

Before configuring data contract rules, create the key encryption key (KEK) that your rules reference. For more information, see Create a KEK.

Configure Confluent Gateway for encryption

To enable encryption, update your gateway.yaml file with the SCHEMA_RULES level and provide the necessary rule executor parameters.

gateway:
  schemaValidation:
    schemaRegistryUrls:
      - "https://psrc-xxxxx.us-east-2.aws.confluent.cloud"
    schemaRegistryConfigs:
      basic.auth.credentials.source: "USER_INFO"
      basic.auth.user.info: "<SR-API-KEY>:<SR-API-SECRET>"
      rule.executors: "encryptField,encryptPayload"                    --- [1]
      rule.executors.encryptField.class: "io.confluent.kafka.schemaregistry.encryption.FieldEncryptionExecutor"
      rule.executors.encryptField.param.secret: "<base64-kms-secret>"  --- [2]
      rule.executors.encryptPayload.class: "io.confluent.kafka.schemaregistry.encryption.EncryptionExecutor"
      rule.executors.encryptPayload.param.secret: "<base64-kms-secret>"
    valueValidationLevel: SCHEMA_RULES                                 --- [3]
    valueSubjectNameStrategy: TOPIC
    topics:
      - name: "payments"                                               --- [4]
        valueSubjectNameStrategy: RECORD                                --- [5]

[1] rule.executors: The executors that Confluent Gateway runs.
- Register encryptField (FieldEncryptionExecutor) for field-level ENCRYPT rules.
- Register encryptPayload (EncryptionExecutor) for full-payload ENCRYPT_PAYLOAD rules.
[2] rule.executors.<name>.param.secret: The base64-encoded secret the key management service (KMS) provider uses to derive encryption keys. In production, provide this through an environment variable or a secret store instead of a plain-text value. For more information, see Create a KEK.
[3] SCHEMA_RULES: Enables round-trip validation and rule execution. Confluent Gateway encrypts data on the produce path and decrypts it on the fetch path.
[4] topics[].name: The name of the Kafka topic to override.
[5] topics[].valueSubjectNameStrategy: Overrides the global subject name strategy for this topic. Use this when a topic, such as payments, carries multiple record types that each need their own data contract rules (for example, different PII tags per record type).

Define the data contract in the schema

Define the encryption logic in the schema ruleSet in Confluent Cloud Schema Registry. Confluent Gateway fetches this definition and applies it to incoming records automatically.

In the following example, Confluent Gateway encrypts only the fields tagged with personally identifiable information (PII) by using a KMS provider. The "schema" value is the JSON-encoded schema definition, in which each sensitive field carries the PII tag. For details on how to tag fields in Avro, JSON Schema, or Protobuf, see Data contracts.

{
  "schema": "...",
  "ruleSet": {
    "domainRules": [{
      "name": "encryptPII",
      "kind": "TRANSFORM",
      "type": "ENCRYPT",
      "mode": "WRITEREAD",
      "tags": ["PII"],
      "params": {
        "encrypt.kek.name": "demo-kek",
        "encrypt.kms.type": "local-kms",
        "encrypt.kms.key.id": "demo-key"
      }
    }]
  }
}

For more information on data contracts and configuring schema ruleSet, see Data contracts.

Encrypt the full payload (optional)

A full-payload rule encrypts the entire record value rather than specific tagged fields. Define an ENCRYPT_PAYLOAD rule in the encodingRules array.

{
  "schema": "...",
  "ruleSet": {
    "encodingRules": [{
      "name": "encryptPayload",
      "kind": "TRANSFORM",
      "type": "ENCRYPT_PAYLOAD",
      "mode": "WRITEREAD",
      "params": {
        "encrypt.kek.name": "demo-kek-payload",
        "encrypt.kms.type": "local-kms",
        "encrypt.kms.key.id": "demo-key"
      }
    }]
  }
}

When the enforcement level is SCHEMA_RULES, Confluent Gateway uses the encryptPayload executor to encrypt the entire serialized value on the produce path and decrypt it on the fetch path.

Disable rule execution on producer applications

To prevent double encryption, disable rule execution on your producer applications. This ensures Confluent Gateway is the single point of enforcement.

In your producer properties, set:

rule.executors._default_.disabled=true

Step 6: Verify centralized governance enforcement

Run a few tests to confirm Confluent Gateway enforces your governance policies. For more tests, error handling, and other validation and encryption examples, see Confluent Gateway examples in GitHub.

Verify schema validation

To verify that Confluent Gateway rejects non-compliant data before it reaches the broker, produce a record that violates the topic’s enforcement level. For example, send a record with a missing schema ID, or use an ID that is not registered under the topic’s subject.

Expected result

Confluent Gateway rejects the record before it reaches the broker and returns an INVALID_RECORD error.

The producer receives an InvalidRecordException message.

Confluent Gateway logs a warning like the following:

[TIMESTAMP] WARN: Schema validation failed for produce request on route=<route>. Returning INVALID_RECORD to producer.

Note

The topic’s enforcement level determines how deeply Confluent Gateway inspects each record:

ID: Validates only the schema ID header.
SCHEMA: Validates the schema ID and deserializes the entire payload, catching malformed records that ID enforcement misses.

Verify encryption

To verify that Confluent Gateway encrypts records on the produce path and decrypts them on the fetch path, produce a plaintext record to a topic configured with SCHEMA_RULES. Consume the record from two separate endpoints to verify the encryption behavior:

Directly from the Kafka broker: The data is encrypted. Field-level rules encrypt the tagged fields, and full-payload rules encrypt the entire message value.
Through Confluent Cloud Gateway: The data is returned as the original plaintext.

Handle errors and monitor governance enforcement

Confluent Gateway returns specific error codes for validation and decryption failures, and logs each failure. Use the following reference to monitor enforcement activity and handle errors.

Error handling and response codes

When Confluent Gateway detects a policy violation or a schema mismatch, it returns specific error codes. This allows client applications to handle failures without losing their connection. For example:

Produce path (validation failure): If a record fails validation, Confluent Gateway returns an INVALID_RECORD error to the producer. The connection remains open, allowing the producer to retry or process the next record in the batch.
Produce path (Schema Registry or KMS unavailable): If Confluent Gateway cannot reach Confluent Cloud Schema Registry or the KMS while validating or encrypting a record, it returns an UNKNOWN_SERVER_ERROR error to the producer instead of INVALID_RECORD, signaling a transient failure rather than a rejected record. The connection remains open, so the producer can retry the record.
Fetch path (decryption failure): If Confluent Gateway cannot decrypt a record (for example, if the KMS is unreachable), it logs an error, sets an UNKNOWN_SERVER_ERROR error code on the affected partition, and returns empty records for that partition. Confluent Gateway does not deliver undecrypted data to the consumer.