<a id="gateway-governance-config"></a>

# 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](gateway-governance.md#gateway-governance).

<a id="gateway-governance-prereqs"></a>

## 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](https://docs.confluent.io/platform/current/installation/license.html#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](https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html).
* **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.

<a id="gateway-governance-sr-auth"></a>

## Step 1: Configure the Schema Registry connection

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

```yaml
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](#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 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](#gateway-governance-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](#gateway-governance-per-topic).

<a id="gateway-governance-enforcement-level"></a>

## 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<br/>requests.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                | Use during initial rollout, or for topics where another system<br/>enforces schema validation.                                                                 |
| Schema ID enforcement  | `ID`                  | **Producer**: Validates that the schema identifier carried within the<br/>record exists under the correct subject in Schema Registry. Confluent Gateway accepts<br/>either a schema GUID in the record header or a numeric schema ID<br/>in the payload, in that order of precedence. Confluent Gateway does not<br/>deserialize the record, so the throughput impact is minimal.<br/><br/>**Consumer**: Pass-through, no validation applied.                                                                                                                                                                                                                        | Use as the default production setting. Confluent Gateway rejects records<br/>with an unknown or mismatched schema identifier before they reach<br/>the broker. |
| Deep schema validation | `SCHEMA`              | **Producer**: Deserializes the payload and validates every field<br/>against the registered schema, then forwards the original bytes to<br/>the broker unchanged. Does not apply data contract rules.<br/><br/>**Consumer**: Pass-through, no validation applied.                                                                                                                                                                                                                                                                                                                                                                                                    | Use for critical ingress pipelines where malformed payloads<br/>could cause downstream failures and `ID` validation alone<br/>is not sufficient.               |
| Data contract rules    | `SCHEMA_RULES`        | **Producer**: Performs a full round-trip on each record. Confluent Gateway<br/>deserializes the payload, then re-serializes it with data contract<br/>rules applied (for example, encrypting fields tagged for<br/>field-level encryption) before forwarding to the broker. For more<br/>details, see<br/>[Step 5: Configure data contract rules and encryption](#gateway-governance-data-contracts).<br/><br/>**Consumer**: Performs a full round-trip on each record. Confluent Gateway<br/>deserializes the payload with data contract rules applied (for<br/>example, decrypting encrypted fields), then re-serializes it<br/>before delivering to the consumer. | Use when you need field-level encryption or data quality rule<br/>enforcement in addition to schema validation.                                                |

Add `valueValidationLevel` to the `schemaValidation` parameters configured
in [Step 1: Configure the Schema Registry connection](#gateway-governance-sr-auth).

```yaml
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`.

<a id="gateway-governance-subject-name-strategy"></a>

## 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.<br/>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<br/>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<br/>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<br/>back to `TOPIC` when no association exists. Use this strategy when you<br/>manage schema-to-topic mappings centrally in Schema Registry. | The subject that Schema Registry associates with the schema.      |

<a id="gateway-governance-per-topic"></a>

## 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.

```yaml
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`.

<a id="gateway-governance-data-contracts"></a>

## 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](https://docs.confluent.io/platform/current/schema-registry/fundamentals/data-contracts.html#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](https://docs.confluent.io/platform/current/security/protect-data/csfle/overview.html).

### 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.

```yaml
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](https://docs.confluent.io/platform/current/schema-registry/fundamentals/data-contracts.html#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](https://docs.confluent.io/platform/current/schema-registry/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/platform/current/schema-registry/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 to ensure that 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 similar to the following:
  ```text
  [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.