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

# Enforce Centralized Governance with Confluent Gateway

To enforce centralized governance on your Apache Kafka® traffic, you must configure
Confluent Gateway to communicate with Confluent Schema Registry and enable governance for record keys
and values 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). To configure governance with Confluent for Kubernetes, see
[Enforce Centralized Governance for Confluent Gateway using Confluent for
Kubernetes](https://docs.confluent.io/operator/current/gateway/co-gateway-governance.html).

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

## Prerequisites

Ensure your environment meets the following requirements:

* **Gateway version:** Deploy Confluent Gateway version 1.4 or later.
* **Licensing:** Encryption rules require the Client-Side Field Level Encryption (CSFLE) add-on
  license. For more information, see [Manage Confluent Platform Licenses](../../installation/license.md#cp-license-overview).
* **Schema Registry:** Deploy Confluent Schema Registry 8.4 or later.
* **Schema format:** Use Confluent Schema Registry serializers that produce the standard
  Confluent wire format, which is a magic byte `0x00` followed by a four-byte schema
  ID, or that carry a schema GUID in a record header.

  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).
* **Gateway authentication:** Provision Confluent Gateway with valid Schema Registry
  credentials using basic authentication or mTLS.
* **Access control:** Grant Confluent Gateway read access to Schema Registry so it can read the
  schemas for all topics that flow through it.
* **Encryption rules (optional):** If you use encryption rules, make sure that
  Confluent Gateway can reach a key management service (KMS). For more information, see
  [Configure the KMS provider](https://docs.confluent.io/platform/current/security/protect-data/csfle/quick-start.html#step-1-configure-the-kms-provider).
  Also set up Schema Registry as described in [Requirements](../../security/protect-data/csfle/client-side.md#csfle-requirements).

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

## Step 1: Configure the Schema Registry connection

Connect Confluent Gateway to Schema Registry so that Confluent Gateway can retrieve schemas and validate
incoming records.

```yaml
gateway:
  policies:
    dataGovernance:
      schemaRegistryUrls:                                  --- [1]
        - "https://schema-registry:8081"
      schemaRegistryConfigs:                               --- [2]
        config.providers: "env"                            --- [3]
        config.providers.env.class: "org.apache.kafka.common.config.provider.EnvVarConfigProvider"
        basic.auth.credentials.source: "USER_INFO"         --- [4]
        basic.auth.user.info: "${env:SR_USER}:${env:SR_PASSWORD}"  --- [5]
      context: null                                        --- [6]
      key:
        enabled: false
      value:
        enabled: true
```

* [1] `schemaRegistryUrls`: A list of one or more Schema Registry URLs. Providing
  multiple URLs enables high availability. If the primary URL is
  unreachable, Confluent Gateway connects to the next URL in the list.
* [2] `schemaRegistryConfigs`: A container for all configurations passed
  to the Schema Registry client, including security protocols, timeouts, and
  authentication details such as basic authentication and mTLS. For mTLS,
  set the `schema.registry.ssl.keystore.*` and
  `schema.registry.ssl.truststore.*` keys.
* [3] `config.providers`: The configuration providers that Confluent Gateway uses
  to resolve `${...}` placeholders in `schemaRegistryConfigs`. Confluent Gateway
  resolves placeholders only in `schemaRegistryConfigs`, not elsewhere in
  the Confluent Gateway configuration. Confluent Gateway leaves placeholders unresolved unless
  you declare a provider. You can use the Kafka built-in providers:
  `EnvVarConfigProvider`, `FileConfigProvider`, and
  `DirectoryConfigProvider`. For more information, see [Configuration
  Providers](https://kafka.apache.org/43/configuration/configuration-providers/).
  This example declares `env` to read values from environment variables.
* [4] `basic.auth.credentials.source`: The source of the credentials for
  basic authentication. Set this to `USER_INFO` to read the username and
  password from the `basic.auth.user.info` property.
* [5] `basic.auth.user.info`: The credentials that Confluent Gateway uses to
  authenticate to Schema Registry, in `username:password` format. In production,
  resolve the values from environment variables or a secrets provider instead
  of using plain text.
* [6] `context`: The Schema Registry context to use for schema lookups. Use a
  context to isolate schemas by environment or tenant. When `context` is
  set to `null` (the default), Confluent Gateway uses the default Schema Registry context.

You must include the `key` and `value` blocks. For more information, see
[Step 2: Enable governance for record keys and values](#gateway-governance-enable).

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

## Step 2: Enable governance for record keys and values

Use the `enabled` setting to enable or disable governance for the record key
and the record value. Confluent Gateway evaluates the record key and value
independently, so you can govern only the one that has a schema. For example,
if your record values use Apache Avro® but your record keys are plain strings,
enable governance for record values only.

```yaml
gateway:
  policies:
    dataGovernance:
      key:
        enabled: false   # record keys are plain strings, no schema
      value:
        enabled: true
```

This example disables governance for record keys, which are plain strings
with no schema, and enables it for record values.

Set `enabled` explicitly for both `key` and `value`. Neither setting has
a default value.

* When `enabled` is `true` for the record key or value, Confluent Gateway performs
  schema ID enforcement and deep schema validation for it on the produce and
  fetch paths. If the schema delegates encryption rules to Confluent Gateway, Confluent Gateway
  also runs them. For more information, see
  [Step 4: Configure data contract rules and encryption](#gateway-governance-data-contracts).
* When `enabled` is `false` for the record key or value, Confluent Gateway forwards
  it unchanged.

For JSON Schema, Confluent Gateway validates the payload against schema constraints,
such as `minimum` and `enum`, only when you set
`json.fail.invalid.schema` to `"true"` in `schemaRegistryConfigs`.

#### NOTE
If your schemas use `enableAt: GATEWAY` and you set `enabled` to
`false`, Confluent Gateway doesn’t encrypt records while governance is off.

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

## Step 3: Configure governance per topic

Add per-topic overrides to change the governance settings for specific topics.
By default, the `key` and `value` settings apply to all topics.

In this example, governance is on for values globally. Confluent Gateway bypasses
governance for the `public-events` topic and uses the `RECORD` subject
name strategy for the `orders` topic.

```yaml
gateway:
  policies:
    dataGovernance:
      schemaRegistryUrls:
        - "https://schema-registry:8081"
      key:
        enabled: false
      value:
        enabled: true                    --- [1]
      topics:
        - name: "public-events"          --- [2]
          value:
            enabled: false               --- [3]
        - name: "orders"                 --- [2]
          value:
            subjectNameStrategy: RECORD  --- [4]
```

* [1] `value.enabled`: The global setting applied to all topics not listed
  under `topics`.
* [2] `topics[].name`: The name of the Kafka topic to override.
* [3] `topics[].value.enabled`: Enables or disables governance for record
  values on this topic. Overrides the global `value.enabled` setting.
* [4] `topics[].value.subjectNameStrategy`: The subject name strategy for
  record values on this topic. Overrides the global
  `value.subjectNameStrategy` setting. Valid values are `TOPIC` (default),
  `RECORD`, and `TOPIC_RECORD`. For how each strategy builds the subject
  name, see [Subject name strategy](https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#subject-name-strategy)
  in the Schema Registry documentation.

Each entry must override at least one setting for `key` or `value`.
Confluent Gateway applies the global value to any setting you omit. Use the `key`
block to override settings for record keys on a topic.

#### NOTE
Per-topic overrides take precedence over the global settings. Setting a
topic’s `enabled` to `false` disables governance for the record key
or value on that topic even when you enable it globally.

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

## Step 4: Configure data contract rules and encryption

Confluent Gateway applies data contract rules defined in Confluent Schema Registry to records on topics
where governance is enabled and the schema delegates the rules to Confluent Gateway.
You typically use these rules for field-level or full-payload encryption, where
Confluent Gateway encrypts or decrypts sensitive data before it reaches the broker.

Before you configure data contract rules, complete the encryption
prerequisites in [Prerequisites](#gateway-governance-prereqs).

### Configure Confluent Gateway for encryption

To configure encryption, enable governance for the record key or value that
contains the sensitive data. Then register the rule executors that run the
Schema Registry encryption rules.

```yaml
gateway:
  policies:
    dataGovernance:
      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>"
      key:
        enabled: false
      value:
        enabled: false                                                   --- [3]
      topics:                                                            --- [4]
        - name: "payments"
          value:
            enabled: true
```

* [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 secrets provider instead of a
  plain-text value. For more information, see [Configure the KMS provider](https://docs.confluent.io/platform/current/security/protect-data/csfle/quick-start.html#step-1-configure-the-kms-provider).
* [3] `value.enabled`: Enables or disables governance for record values for
  all topics. This example sets it to `false` so that governance applies
  only to the topics you override in [4]. Encryption rules run only for record
  keys and values that have governance enabled.
* [4] `topics`: (Optional) The per-topic overrides. Use overrides to limit
  governance to topics that carry sensitive data. This example enables
  governance for record values on the `payments` topic only.

If the KEK uses a cloud key management service (KMS), such as AWS KMS,
Azure Key Vault, or Google Cloud KMS, also set the credentials that Confluent Gateway uses to
access the KMS as `rule.executors.<executor_name>.param.<setting>`
properties in `schemaRegistryConfigs`. For the settings of each KMS, see
[Configure Client-Side Field Level Encryption on Confluent Platform](../../security/protect-data/csfle/config-parameters.md#config-parameters-csfle).

### Choose where encryption runs

Encryption rules in a schema’s rule set include an `enableAt` setting.
Confluent Gateway runs the rules only when `enableAt` is `GATEWAY` or `ALL`, or
isn’t set. To have Confluent Gateway manage encryption, set `enableAt` to
`GATEWAY`.

### Define the data contract in the schema

Define the encryption logic in the schema `ruleSet` in Schema Registry. Set
`enableAt` to `GATEWAY`. 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"
      }
    }],
    "enableAt": "GATEWAY"
  }
}
```

On the produce path, Confluent Gateway encrypts the tagged fields before writing them
to Kafka. On the fetch path, Confluent Gateway decrypts the tagged fields. Untagged
fields are unchanged.

### 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. Register the `encryptPayload` executor in Confluent Gateway.

```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"
      }
    }],
    "enableAt": "GATEWAY"
  }
}
```

Confluent Gateway encrypts the entire serialized value on the produce path and
decrypts it on the fetch path.

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

## Step 5: Verify centralized governance enforcement

Run a few tests to confirm Confluent Gateway enforces your governance policies.

### Verify schema validation

To verify that Confluent Gateway rejects noncompliant data before it reaches the
broker, produce a record that violates the schema. For example, send a
record with a missing schema ID. Or, send a record with a valid schema ID
and a payload that doesn’t conform to the schema.

**Expected result**

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

* The producer receives an `InvalidRecordException`.
* Confluent Gateway logs a message similar to the following:
  ```text
  Schema validation failed; rejecting record
  ```

### Verify encryption

To verify that Confluent Gateway encrypts records on the produce path and decrypts
them on the fetch path, use a topic whose schema has an encryption rule with
`enableAt` set to `GATEWAY`. Produce a plaintext record to that topic.
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 record
  value.
* **Through Confluent Gateway**: Confluent Gateway returns the original plaintext.

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

## Handle errors and monitor governance enforcement

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

### Error handling and response codes

When Confluent Gateway detects a policy violation or a schema mismatch, it returns
specific error codes. These error codes let client applications 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 and rejects the entire produce request, not only the failing
  record. The connection remains open.
* **Produce path, Schema Registry or KMS unavailable:**
  If Confluent Gateway can’t reach Schema Registry or the KMS, or Schema Registry rejects the credentials,
  it returns a `REQUEST_TIMED_OUT` error, and the producer can retry the
  request.
* **Produce path, unexpected error:**
  If Confluent Gateway encounters an unexpected error, such as an unexpected response
  from Schema Registry, it returns an `UNKNOWN_SERVER_ERROR` error to the producer and
  rejects the entire produce request.
* **Fetch path, any failure:**
  If Confluent Gateway can’t validate or decrypt a record, for example, because the
  KMS is unreachable, it logs an error. Confluent Gateway then sets an
  `UNKNOWN_SERVER_ERROR` code on the affected partition and returns empty
  records for that partition. Confluent Gateway returns the other partitions normally
  in the same fetch response and doesn’t deliver data that failed validation
  or decryption to the consumer.

### Metrics

Confluent Gateway records the `gateway.governance.data.latency` timer for each
partition it validates. The timer’s count is the number of validated
partitions. The timer has the following tags:

* `route`: The Confluent Gateway route name.
* `topic`: The Kafka topic name.
* `mode`: `produce` or `fetch`.
* `outcome`: `success` or `rejected`.
* `failure.origin`: The origin of the failure, or `none` on success.
  Prometheus renders this tag as `failure_origin`.

Confluent Gateway increments the `gateway.governance.data.skipped` counter, tagged
with `route`, `topic`, and `mode`, for partitions that need no
validation. A rise in this counter for a topic that you expect to be governed
indicates that governance is effectively off for that topic.

When a schema carries rules, such as field encryption, Confluent Gateway also publishes
the following per-rule metrics, tagged with `rule_type`, `rule_mode`,
`subject`, and `rule_name`:

* `gateway.governance.data.rule.executions.total`
* `gateway.governance.data.rule.success.total`
* `gateway.governance.data.rule.failure.total`
* `gateway.governance.data.rule.skipped.total`
* `gateway.governance.data.rule.actions.total`
* `gateway.governance.data.rule.last.success.timestamp.ms`