<a id="flink-csfle"></a>

# Use Client-Side Field Level Encryption with Confluent Cloud for Apache Flink

Protect sensitive fields in Apache Kafka® messages and process them in
Confluent Cloud for Apache Flink® by using
[client-side field level encryption](../../security/encrypt/csfle/client-side.md#use-client-side-field-level-encryption)
(CSFLE). CSFLE encrypts individual fields on the producer before the data
reaches the broker, and Flink transparently decrypts those fields at query
time so they remain protected even if the server is compromised. This
walkthrough creates a schema with encrypted fields, sets up an Flink table
that reads the encrypted topic, and runs queries that decrypt the data for
processing.

#### NOTE
Client-side field level encryption with Flink is a Limited Availability
feature in Confluent Cloud.

For limitations during the Limited Availability phase, see
[Known limitations and caveats](../../security/encrypt/csfle/flink-integration.md#csfle-flink-integration-limitations).

If you want to participate in the Limited Availability Program, contact
your Confluent account team.

#### IMPORTANT
If you do not share access to your Key Encryption Key (KEK) with
Confluent, do not grant service accounts for Flink queries RBAC access
on your KEK.

For more information on Flink support for CSFLE, see
[Process Encrypted Data with Confluent Cloud for Apache Flink](../../security/encrypt/csfle/flink-integration.md#csfle-flink-integration).

Confluent Cloud for Apache Flink also supports client-side payload encryption (CSPE), which encrypts
the entire message payload. For more information on using CSPE, see
[Use Client-Side Payload Encryption](../../security/encrypt/cspe.md#use-client-side-payload-encryption).

## Prerequisites

- Access to Confluent Cloud
- An environment with the Stream Governance Advanced package enabled
- A provisioned Flink compute pool.
- Authorized user with the following roles for KEK access:
  - DeveloperRead on the input topic.
  - DeveloperWrite on the output topic.
  - DeveloperRead on Schema Registry subjects for input and output topics.
  - DeveloperWrite on the KEK for the first write to generate the Data
    Encryption Key.
  - DeveloperRead on the KEK for subsequent reads and writes.
  - DeveloperManage on the output topic and subject for creating output
    tables.
- The ARN for an encryption key generated in AWS KMS to use as the KEK for
  encryption rules.

## Step 1: Set up CSFLE on Confluent Cloud

Register the KEK in Confluent Cloud.

1. Log in to Confluent Cloud Console.
2. Navigate to the environment where you want to use CSFLE.
3. In the navigation menu, navigate to **Schema Registry > Encryption Keys**
   and click **Add encryption key**.
4. Name the key “csfle-key”.
5. In the **Key management system provider** dropdown, select **AWS**.
6. In the **Amazon resource name (key ID)** textbox, provide the ARN of the AWS
   KMS key.
7. Enable **Share key access with Confluent Cloud** to give Flink access to the
   KEK.
8. Click **Add** to save the key.

## Step 2: Create a topic and schema with encrypted fields

Create a Flink table for encrypted customer data based on the built-in
`examples.marketplace.customers` table, and tag sensitive fields as PII.

1. Create a table for encrypted customer data:
   ```sql
   CREATE TABLE customers_encrypted (
     customer_id INT NOT NULL,
     customer_name VARCHAR,
     address STRING,
     postcode STRING,
     city STRING,
     email STRING,
     PRIMARY KEY (customer_id) NOT ENFORCED
   );
   ```

   This statement creates the `customers_encrypted` topic and registers its
   Avro schema in Schema Registry.
2. In Confluent Cloud Console, navigate to **Stream Governance > Tags** and verify
   that a `PII` tag exists. If it doesn’t, click **Add tag** and create a tag
   named `PII`.
3. Navigate to **Schema Registry > Data contracts** and open the
   `customers_encrypted-value` subject.
4. Click **Evolve**, and in the editor, add the `PII` tag to the
   `email` and `address` fields. These tags identify the
   fields that encryption rules protect.
   ```json
   {
     "fields": [
         {
         "default": null,
         "name": "customer_id",
         "type": [
             "null",
             "int"
         ]
         },
         {
         "default": null,
         "name": "customer_name",
         "type": [
             "null",
             "string"
         ]
         },
         {
         "default": null,
         "name": "address",
         "type": [
             "null",
             "string"
         ],
         "confluent:tags": ["PII"]
         },
         {
         "default": null,
         "name": "postcode",
         "type": [
             "null",
             "string"
         ]
         },
         {
         "default": null,
         "name": "city",
         "type": [
             "null",
             "string"
         ]
         },
         {
         "default": null,
         "name": "email",
         "type": [
             "null",
             "string"
         ],
         "confluent:tags": ["PII"]
         }
     ],
     "name": "customers_encrypted_value",
     "namespace": "org.apache.flink.avro.generated.record",
     "type": "record"
   }
   ```
5. Before the closing curly brace, add the following JSON to attach a domain
   rule that encrypts fields tagged with `PII` using the KEK you registered
   in Step 1. Ensure that you add a comma after the `"type": "record"` field
   in the previous snippet to maintain valid JSON syntax.

   For client-side payload encryption (CSPE), specify
   `"type": "ENCRYPT_PAYLOAD"`.
   ```json
   "domainRules": [
     {
       "name": "encrypt-pii",
       "kind": "TRANSFORM",
       "mode": "WRITEREAD",
       "type": "ENCRYPT",
       "tags": ["PII"],
       "params": {
         "encrypt.kek.name": "csfle-key"
       },
       "onFailure": "ERROR",
       "onSuccess": "NONE"
     }
   ]
   ```
6. Click **Save** to update the schema with the PII tags and domain rules.
7. Return to your workspace and run the following query to populate the table
   with data from the `customers` table:
   ```sql
   INSERT INTO customers_encrypted
     SELECT *
     FROM `examples`.`marketplace`.`customers`;
   ```

   This statement reads from the built-in `customers` table. Flink
   automatically encrypts the `email` and `address` fields according
   to the domain rules you attached in the previous step.

## Step 3: Query encrypted data

Run queries on the encrypted data. If you have DeveloperRead permission on the
KEK, Flink transparently decrypts the encrypted fields for processing.

**Example: Count customers by email domain**

```sql
SELECT
  SUBSTRING(email, POSITION('@' IN email) + 1) AS domain,
  COUNT(*) AS customer_count
FROM customers_encrypted
GROUP BY SUBSTRING(email, POSITION('@' IN email) + 1);
```

In this query:

- Flink decrypts the `email` field.
- The SQL operations work with plaintext email addresses.
- The results show aggregated counts by email domain.

**Example: Filter by encrypted field**

```sql
SELECT customer_id, email, city
FROM customers_encrypted
WHERE address LIKE '%New York%';
```

In this query:

- Flink decrypts the `address` field for filtering.
- The `WHERE` clause operates on plaintext address values.
- Flink returns only matching rows.

## What happens without key permissions

If you run Flink SQL statements without DeveloperRead permission on the
encryption keys, the behavior depends on the encryption type:

CSFLE without permissions
: Encrypted fields remain encrypted and pass through without decryption.
  Operations like `COUNT(DISTINCT address)` count distinct ciphertext
  values, not distinct addresses. For more information, see
  [Process Encrypted Data with Confluent Cloud for Apache Flink](../../security/encrypt/csfle/flink-integration.md#csfle-flink-integration).

CSPE without permissions
: The Flink SQL statement fails because Flink cannot decrypt the payload.

For a complete description of permission-based behavior, see
[Process Encrypted Data with Confluent Cloud for Apache Flink](../../security/encrypt/csfle/flink-integration.md#csfle-flink-integration).

## Related content

- [Process Encrypted Data with Confluent Cloud for Apache Flink](../../security/encrypt/csfle/flink-integration.md#csfle-flink-integration)
- [Use Client-Side Field Level Encryption](../../security/encrypt/csfle/client-side.md#use-client-side-field-level-encryption)
- [Use Client-Side Payload Encryption](../../security/encrypt/cspe.md#use-client-side-payload-encryption)
- [Manage Client-Side Field Level Encryption using Confluent Cloud Console](../../security/encrypt/csfle/manage-csfle.md#manage-csfle-ccloud-console)
- [Manage Encryption Keys for CSFLE](../../security/encrypt/csfle/manage-keys.md#manage-encryption-keys-csfle)
- [Code Examples for Client-Side Field Level Encryption in Confluent Cloud](../../security/encrypt/csfle/code-examples.md#code-examples-csfle)
- Confluent Developer Course: [Client-Side Field-Level Encryption (CSFLE)](https://developer.confluent.io/courses/csfle/csfle-overview/)

#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).