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. To configure governance with Confluent for Kubernetes, see Enforce Centralized Governance for Confluent Gateway using Confluent for Kubernetes.

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.
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.
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. Also set up Schema Registry as described in Requirements.

Step 1: Configure the Schema Registry connection

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

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

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.

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

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.

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

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.

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.

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

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.

{
  "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.

{
  "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.

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

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