Enforce Centralized Governance with Confluent Gateway

To enforce centralized governance on your Kafka traffic, you must configure Confluent Gateway to communicate with an on-premises Confluent Schema Registry and set the enforcement level 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.
Licensing: Get the Confluent Enterprise License for Confluent Private Cloud subscription that Schema Registry requires. For more information, see Confluent Enterprise License for CPC subscription.
Schema format: Deploy Confluent Schema Registry serializers that produce the standard Confluent wire format (a magic byte 0x00 followed by a 4-byte schema ID).
Confluent Gateway validates records serialized with Apache Avro®, JSON Schema, or Protobuf. For more information, see Serializer and Deserializer Development.
Schema Registry: Deploy an on-premises Confluent Schema Registry.
Gateway authentication: Provision Confluent Gateway with valid Schema Registry credentials using basic authentication.
Access control: Grant Confluent Gateway read access to Schema Registry so it can read the schemas for all topics that flow through it.

Step 1: Configure the Schema Registry connection

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

gateway:
  schemaValidation:
    schemaRegistryUrls:                                  --- [1]
      - "https://schema-registry:8081"
    schemaRegistryConfigs:                               --- [2]
      basic.auth.credentials.source: "USER_INFO"         --- [3]
      basic.auth.user.info: "${SR_USER}:${SR_PASSWORD}"  --- [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 Schema Registry URLs. Providing multiple URLs enables high availability. Confluent Gateway connects to the next URL in the list if the primary URL is unreachable.
[2] schemaRegistryConfigs: A container for all configurations passed to the Schema Registry client, including security protocols, timeouts, and authentication details such as basic authentication, OAuth, OIDC, and mTLS.
[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 credentials used to authenticate to Schema Registry in username:password format. For production environments, use environment variables or a secrets provider rather than plain text.
[5] context: The 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 Schema Registry context.
[6] valueValidationLevel: The enforcement level applied to record values on the produce path. Defaults to NONE. For more information on enforcement levels, 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 Schema Registry.
[8] valueSubjectNameStrategy: The subject name strategy that Confluent Gateway uses to derive the subject name for schema lookup in 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 Schema Registry. Set this property only if your producers serialize keys with a schema registered in Schema Registry.
[10] topics: (Optional) A list of per-topic overrides. Each entry can override validation levels, subject name strategies, and the 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 identifier carried within the record exists under the correct subject in Schema Registry. Confluent Gateway accepts either a schema GUID in the record header or a numeric schema ID in the payload, in that order of precedence. 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 identifier 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 a full round-trip on each record. Confluent Gateway deserializes the payload, then re-serializes it with data contract rules applied (for example, encrypting fields tagged for field-level encryption) before forwarding to the broker. For more details, see Step 5: Configure data contract rules and encryption. Consumer: Performs a full round-trip on each record. Confluent Gateway deserializes the payload with data contract rules applied (for example, decrypting encrypted fields), then re-serializes it before delivering to the consumer.	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 Schema Registry 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 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 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 Schema Registry.	The subject that 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 applies to all topics.

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

gateway:
  schemaValidation:
    schemaRegistryUrls:
      - "https://schema-registry:8081"
    schemaRegistryConfigs:
      basic.auth.credentials.source: "USER_INFO"
      basic.auth.user.info: "${SR_USER}:${SR_PASSWORD}"
    keyValidationLevel: NONE
    valueValidationLevel: ID          --- [1]
    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 topic applied to record values. Overrides the global valueValidationLevel. Accepts the same values as discussed in Step 2: NONE, ID, SCHEMA, SCHEMA_RULES.
Use keyValidationLevel to set a different enforcement level for record keys on a per-topic basis.

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 Schema Registry to the incoming records. These rules are commonly used for field-level or full-payload encryption, where Confluent Gateway encrypts or decrypts sensitive data before it reaches the broker.

Prerequisites

Before you configure data contract rules:

Ensure that a Key Management Service (KMS) is reachable by Confluent Gateway. For more information, see KMS configurations.
Configure Schema Registry to load the RuleSetResourceExtension and DekRegistryResourceExtension resource extensions, which store the ruleSet and manage data encryption keys (DEKs). Without these extensions, Schema Registry discards the ruleSet, and Confluent Gateway cannot encrypt or decrypt records.
Enable the Confluent Enterprise license for Confluent Private Cloud subscription on Schema Registry. Schema Registry requires this license to load the RuleSetResourceExtension and DekRegistryResourceExtension extensions. Set confluent.license.addon.csfle to the same value as your confluent.license key. For more information, see Client-Side Field Level Encryption.

Configure Confluent Gateway for encryption

To enable encryption, set valueValidationLevel to SCHEMA_RULES and register the rule executors that run the Schema Registry encryption rules.

gateway:
  schemaValidation:
    schemaRegistryConfigs:
      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]
    topics:                                                            --- [4]
      - name: "payments"
        valueValidationLevel: SCHEMA_RULES

[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 local 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 KMS configurations.
[3] valueValidationLevel: The enforcement level for the route. If you set this to SCHEMA_RULES, Confluent Gateway runs a full round trip on each record, encrypting on the produce path and decrypting on the fetch path.
[4] topics: The per-topic overrides. Apply SCHEMA_RULES only to topics that carry sensitive data.

Define the data contract in the schema

Define the encryption logic in the schema ruleSet in 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 local 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 to ensure that 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 similar to 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 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 governance 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.
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.
Fetch path (other validation failure): If a record fails validation on the fetch path for a reason other than decryption, Confluent Gateway passes the original records through unchanged and logs a warning. Because the data is already committed to the broker, this prevents a single validation error from halting consumer progress.