# 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](gateway-governance.md#gateway-governance). ## 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](https://docs.confluent.io/cloud/current/sr/fundamentals/serdes-develop/index.html). * **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. ```yaml 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: ":" --- [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 `:` 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](#gateway-governance-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](#gateway-governance-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](#gateway-governance-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](#gateway-governance-data-contracts).

**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-governance-sr-auth). ```yaml gateway: schemaValidation: valueValidationLevel: ``` `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 `-value` (or `-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 `-`. 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](#gateway-governance-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. ```yaml gateway: schemaValidation: schemaRegistryUrls: - "https://psrc-xxxxx.us-east-2.aws.confluent.cloud" schemaRegistryConfigs: basic.auth.credentials.source: "USER_INFO" basic.auth.user.info: ":" 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](#gateway-governance-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](../../../security/encrypt/csfle/manage-keys.md#create-kek-csfle). ### 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. ```yaml gateway: schemaValidation: schemaRegistryUrls: - "https://psrc-xxxxx.us-east-2.aws.confluent.cloud" schemaRegistryConfigs: basic.auth.credentials.source: "USER_INFO" basic.auth.user.info: ":" rule.executors: "encryptField,encryptPayload" --- [1] rule.executors.encryptField.class: "io.confluent.kafka.schemaregistry.encryption.FieldEncryptionExecutor" rule.executors.encryptField.param.secret: "" --- [2] rule.executors.encryptPayload.class: "io.confluent.kafka.schemaregistry.encryption.EncryptionExecutor" rule.executors.encryptPayload.param.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..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](../../../security/encrypt/csfle/manage-keys.md#create-kek-csfle). * [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](https://docs.confluent.io/cloud/current/sr/fundamentals/data-contracts.html). ```json { "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](https://docs.confluent.io/cloud/current/sr/fundamentals/data-contracts.html). ### 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. ```json { "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: ```text 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](https://github.com/confluentinc/gateway-images/tree/master/examples/governance). ### 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: ```text [TIMESTAMP] WARN: Schema validation failed for produce request on 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.