Enforce Centralized Governance for Confluent Gateway using Confluent for Kubernetes

Configure Confluent Gateway deployed with Confluent for Kubernetes (CFK) to enforce Confluent Schema Registry data contracts on the traffic between your Apache Kafka® clients and clusters. You define policies once in Schema Registry, and Confluent Gateway enforces them on every producer and consumer request without changes to your client applications.

For an overview of how centralized governance enforcement works, including the supported checks and limitations, see the following topics:

For Confluent Private Cloud Gateway, see Enforce Data Governance Centrally with Confluent Gateway.
For the Confluent Gateway add-on for Confluent Platform, see Enforce Data Governance Centrally with Confluent Gateway.

Before you begin

Ensure your environment meets the following requirements:

Confluent Gateway version 1.4 or later, deployed as described in Configure and Deploy Confluent Gateway using Confluent for Kubernetes.
CFK version 3.4 or later.
For the Confluent Gateway add-on for Confluent Platform, a Confluent Enterprise license for Confluent Platform that includes the Confluent Gateway add-on. To use encryption rules, you also need the Client-Side Field Level Encryption (CSFLE) add-on license, which is a separate purchase on Confluent Platform. For more information, see Confluent Platform licenses.
Confluent Schema Registry on Confluent Cloud, or Confluent Platform Schema Registry 8.4 or later.
Client applications that produce and consume records through Confluent Gateway using Schema Registry serializers. Records must carry a schema ID, either in the Schema Registry wire format or as a schema globally unique identifier (GUID) in a record header.
Network access from Confluent Gateway to Schema Registry and credentials for one of the supported authentication methods. For more information, see Connect to Schema Registry.
For encryption, a key encryption key (KEK) registered in Schema Registry, Schema Registry configured to store encryption rules (the RuleSetResourceExtension and DekRegistryResourceExtension extensions and the client-side field level encryption license add-on), and permission for Confluent Gateway to use the key in your key management service (KMS). For more information, see Configure encryption.

Configure governance

Configure governance in the Gateway custom resource (CR) under spec.policies.dataGovernance. The policy applies to traffic on all Confluent Gateway routes.

apiVersion: platform.confluent.io/v1beta1
kind: Gateway
metadata:
  name: confluent-gateway
  namespace: confluent
spec:
  policies:
    dataGovernance:
      schemaRegistry:
        urls:                      --- [1]
          - "https://<schema-registry-host>:8081"
        authentication: { ... }    --- [2]
        tls: { ... }               --- [3]
        ruleExecutors: [ ... ]     --- [4]
      key:
        enabled: false             --- [5]
      value:
        enabled: true              --- [5]
        subjectNameStrategy: TOPIC --- [6]

[1] schemaRegistry.urls: The Schema Registry endpoints that Confluent Gateway uses to look up schemas and rules. Specify 1 to 16 URLs.
[2] schemaRegistry.authentication.type: How Confluent Gateway authenticates to Schema Registry. Valid values are basic and mtls. For more information, see Connect to Schema Registry.
[3] schemaRegistry.tls: The TLS settings for the Schema Registry connection. For more information, see Connect to Schema Registry.
[4] schemaRegistry.ruleExecutors: The rule executors that run encryption rules. Required only for encryption. For more information, see Configure encryption.
[5] key.enabled and value.enabled: Enable or disable governance for record keys and record values. Both settings are required. When set to false, Confluent Gateway forwards that part of the record unchanged. Confluent Gateway evaluates keys and values independently, so you can govern only the part that has a schema.
[6] key.subjectNameStrategy and value.subjectNameStrategy: How Confluent Gateway maps a record to a Schema Registry subject. Valid values are TOPIC (default), RECORD, and TOPIC_RECORD. Keys and values can use different strategies.

To pass additional Schema Registry client properties, such as timeouts, use schemaRegistry.additionalConfigs.

Note

For JSON Schema, to have Confluent Gateway validate the payload against schema constraints, such as minimum and enum, set json.fail.invalid.schema to "true" in schemaRegistry.additionalConfigs.

Confluent Gateway reads schemas and rules from Schema Registry. You define validation constraints and encryption rules in the schemas, not in the Confluent Gateway CR.

Connect to Schema Registry

Store Schema Registry credentials and certificates in Kubernetes secrets in the Confluent Gateway namespace, and reference them from schemaRegistry.

Confluent Gateway supports basic and mutual TLS (mTLS) authentication to Schema Registry. Use mTLS with Confluent Platform Schema Registry. Schema Registry on Confluent Cloud authenticates clients with API keys, so use basic authentication.

Basic authentication

Create a secret with a basic.txt key:

apiVersion: v1
kind: Secret
metadata:
  name: sr-credentials
  namespace: confluent
type: Opaque
stringData:
  basic.txt: |
    username=<sr-username>
    password=<sr-password>

Reference the secret from the Confluent Gateway CR:

spec:
  policies:
    dataGovernance:
      schemaRegistry:
        urls:
          - "http://schemaregistry.confluent.svc.cluster.local:8081"
        authentication:
          type: basic
          basic:
            secretRef: sr-credentials

If Schema Registry uses a certificate signed by a private certificate authority (CA), also create a secret with a truststore that contains the CA certificate:

apiVersion: v1
kind: Secret
metadata:
  name: sr-truststore
  namespace: confluent
type: Opaque
stringData:
  jksPassword.txt: jksPassword=<truststore-password>
data:
  truststore.jks: <base64-encoded-truststore>

Enable TLS, reference the truststore, and use an https:// URL. If Schema Registry uses a publicly trusted certificate, omit secretRef:

spec:
  policies:
    dataGovernance:
      schemaRegistry:
        urls:
          - "https://schemaregistry.confluent.svc.cluster.local:8081"
        authentication:
          type: basic
          basic:
            secretRef: sr-credentials
        tls:
          enabled: true
          secretRef: sr-truststore

Mutual TLS (mTLS)

Create a single secret with the Confluent Gateway client keystore and the truststore:

apiVersion: v1
kind: Secret
metadata:
  name: sr-tls
  namespace: confluent
type: Opaque
stringData:
  jksPassword.txt: jksPassword=<store-password>
data:
  keystore.jks: <base64-encoded-keystore>
  truststore.jks: <base64-encoded-truststore>

You don’t need a credentials secret. Set the authentication type to mtls and enable TLS:

spec:
  policies:
    dataGovernance:
      schemaRegistry:
        urls:
          - "https://schemaregistry.confluent.svc.cluster.local:8081"
        authentication:
          type: mtls
        tls:
          enabled: true
          secretRef: sr-tls

Configure governance per topic

Use topics to override the governance settings for specific topics. Each entry must set at least one setting under key or value, and Confluent Gateway applies the global value to any setting you omit. You can add up to 200 topics.

In the following example, governance is enabled globally for values, but Confluent Gateway bypasses governance for the public-events topic and uses the RECORD subject name strategy for the inventory topic:

spec:
  policies:
    dataGovernance:
      key:
        enabled: false
      value:
        enabled: true
      topics:
        - name: public-events
          value:
            enabled: false
        - name: inventory
          value:
            subjectNameStrategy: RECORD

Configure encryption

To encrypt and decrypt records on Confluent Gateway, complete the following steps:

Configure Schema Registry to store encryption rules and keys. For more information, see Configure data contract rules and encryption for Confluent Private Cloud Gateway or Configure data contract rules and encryption for the Confluent Gateway add-on for Confluent Platform.
Add rule executors and give Confluent Gateway access to the KMS, as described in Configure KMS access.
Add an encryption rule with enableAt: GATEWAY to the schema, as described in Define encryption rules in the schema.
Enable governance for the key or value that contains the data to encrypt.

Configure KMS access

To run encryption rules, Confluent Gateway needs a rule executor and permission to use the KMS key. Register FieldEncryptionExecutor for field-level encryption rules, and EncryptionExecutor for full-payload encryption rules.

The following example uses AWS KMS on Amazon EKS with IAM Roles for Service Accounts (IRSA):

Annotate the Confluent Gateway service account with an IAM role:

apiVersion: v1
kind: ServiceAccount
metadata:
  name: default
  namespace: confluent
  annotations:
    eks.amazonaws.com/role-arn: arn:aws:iam::<aws-account-id>:role/<irsa-role-name>

Confluent Gateway assumes the role that has access to the KMS key when it calls AWS KMS. Create a secret with the Amazon Resource Name (ARN) of that role:

apiVersion: v1
kind: Secret
metadata:
  name: aws-kms-role-params
  namespace: confluent
type: Opaque
stringData:
  role.arn: arn:aws:iam::<aws-account-id>:role/<kms-access-role-name>

Add the rule executors to the Schema Registry configuration:

spec:
  policies:
    dataGovernance:
      schemaRegistry:
        urls:
          - "http://schemaregistry.confluent.svc.cluster.local:8081"
        authentication:
          type: basic
          basic:
            secretRef: sr-credentials
        ruleExecutors:
          - name: encryptField
            class: io.confluent.kafka.schemaregistry.encryption.FieldEncryptionExecutor
            secretRef: aws-kms-role-params
          - name: encryptPayload
            class: io.confluent.kafka.schemaregistry.encryption.EncryptionExecutor
            secretRef: aws-kms-role-params

The preceding example annotates the default service account. If you set a different service account in spec.podTemplate.serviceAccountName, annotate that one instead.

Grant the IRSA role permission to assume the KMS access role, and grant the KMS access role permission to encrypt and decrypt with the KMS key that your KEK references.

Confluent Gateway also supports Azure Key Vault, Google Cloud KMS, and HashiCorp Vault. To set up a KEK and KMS access for those providers, see Configure the KMS provider.

Define encryption rules in the schema

A schema’s rule set includes an enableAt setting that determines which component runs its encryption rules. Set enableAt to GATEWAY so that only Confluent Gateway runs the rules. For all valid values, see Choose where encryption runs for Confluent Private Cloud Gateway or Choose where encryption runs for the Confluent Gateway add-on for Confluent Platform.

To encrypt tagged fields, add a field-level rule. In the following example, Confluent Gateway encrypts the fields tagged PII:

{
  "schema": "...",
  "ruleSet": {
    "domainRules": [ {
      "name": "encryptPII",
      "kind": "TRANSFORM",
      "type": "ENCRYPT",
      "mode": "WRITEREAD",
      "tags": ["PII"],
      "params": {
        "encrypt.kek.name": "my-kek",
        "encrypt.kms.type": "aws-kms",
        "encrypt.kms.key.id": "<kms-key-arn>"
      },
      "onFailure": "ERROR,NONE"
    } ],
    "enableAt": "GATEWAY"
  }
}

With onFailure set to ERROR,NONE, Confluent Gateway raises an error if encryption fails on the produce path. If decryption fails on the fetch path, Confluent Gateway returns the record with the fields still encrypted and doesn’t raise an error.

To encrypt the entire record value, add a payload rule:

{
  "schema": "...",
  "ruleSet": {
    "encodingRules": [ {
      "name": "encryptPayload",
      "kind": "TRANSFORM",
      "mode": "WRITEREAD",
      "type": "ENCRYPT_PAYLOAD",
      "params": { "encrypt.kek.name": "my-kek" },
      "onSuccess": "NONE",
      "onFailure": "ERROR"
    } ],
    "enableAt": "GATEWAY"
  }
}

On the produce path, Confluent Gateway encrypts the tagged fields or the full payload before writing to Kafka. On the fetch path, it decrypts them for consumers.

To keep client-side encryption while Confluent Gateway validates records, set enableAt to CLIENT, and set execution.environment to CLIENT in your client serializer configuration. If you don’t set enableAt and Confluent Gateway has rule executors configured, both Confluent Gateway and client-side serializers run the rules.

Verify governance

To verify that Confluent Gateway enforces governance, register a schema for a topic, enable governance for values, and produce records through the Confluent Gateway route endpoint.

Before you start, set the SR_URL, SR_USER, SR_PASSWORD, and GATEWAY_BOOTSTRAP environment variables, and create a client.properties file with the security settings that your clients use for the Confluent Gateway route. For more information, see Reconfigure clients.

Register an Apache Avro® schema for the orders topic value:

curl -s -X POST \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -u "$SR_USER:$SR_PASSWORD" \
  "$SR_URL/subjects/orders-value/versions" \
  --data '{"schemaType":"AVRO","schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"com.example\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"currency\",\"type\":\"string\"}]}"}'

Enable governance for values in the Confluent Gateway CR:

spec:
  policies:
    dataGovernance:
      schemaRegistry: { ... }
      key:
        enabled: false
      value:
        enabled: true

Produce a record with the Schema Registry serializer:

kafka-avro-console-producer \
  --bootstrap-server "$GATEWAY_BOOTSTRAP" \
  --topic orders \
  --producer.config client.properties \
  --property schema.registry.url="$SR_URL" \
  --property basic.auth.credentials.source=USER_INFO \
  --property basic.auth.user.info="$SR_USER:$SR_PASSWORD" \
  --property value.schema.id=<orders-value-schema-id>
> {"order_id":"A-1001","amount":42.5,"currency":"USD"}

Confluent Gateway accepts the record.

Produce a plain-text record, which has no schema ID:

kafka-console-producer \
  --bootstrap-server "$GATEWAY_BOOTSTRAP" \
  --producer.config client.properties \
  --topic orders
> {"order_id":"A-1002","amount":10.0,"currency":"USD"}

Confluent Gateway rejects the record, and the producer receives an INVALID_RECORD error.

For more information about the errors that Confluent Gateway returns, see Error handling and response codes for Confluent Private Cloud Gateway or Error handling and response codes for the Confluent Gateway add-on for Confluent Platform.