Manage Client-Side Encryption in Confluent Cloud for Self-Managed Connectors

Client-Side Field Level Encryption (CSFLE) and client-side payload encryption (CSPE) are security features that allow you to safeguard sensitive data, such as personally identifiable information (PII), by enabling field-level or payload encryption both at the producer and consumer levels. By encrypting and decrypting individual fields or complete payload/message within your data, CSFLE or CSPE ensures that access to sensitive information is tightly controlled, granting only authorized stakeholders access to the data they are permitted to see.

Important

  • CSFLE or CSPE is supported for self-managed connectors only when both Schema Registry and Kafka are running in Confluent Cloud. This is essential for CSFLE or CSPE functionality when using self-managed connectors.

  • Confluent Platform 7.5.4 or later, or Confluent Platform 7.6.1 or later, is required to use CSFLE or CSPE with self-managed connectors when Schema Registry and Kafka are running in Confluent Cloud.

  • CSLFE or CSPE for self-managed connectors when Schema Registry and Kafka are both running on Confluent Platform is only supported on Confluent Platform 8.0 and later. This is not supported on Confluent Platform versions earlier than 8.0.

Limitations

Refer to the following for usage limitations:

  • The connector does not support automatic schema registration. You must manually register schemas before creating the connectors.

  • The connector only supports encryption for fields of type string or bytes for CSFLE.

  • CSFLE does not support string and bytes fields within nested JSON_SR formats when the value.converter.decimal.format is set to BASE64. To workaround this limitation, set value.converter.decimal.format to NUMERIC.

Note

The reporter topics are not covered. Ensure that the error and success response do not contain any sensitive information while using reporter topics.

Supported connectors

The following table list the connector and its minimum version that support CSFLE.

Connector

Minimum supported version

ActiveMQ Source

12.2.9

Amazon CloudWatch Metrics Sink

2.0.1

Amazon DynamoDB Sink

1.4.1

Amazon Kinesis Source

1.3.27

Amazon Redshift Sink

1.2.6

Amazon S3 Sink

10.6.0

Amazon S3 Source

2.6.10

Amazon SQS Source

2.0.3

Apache Kudu (Source and Sink)

1.0.5

AWS Lambda Sink

2.0.10

Azure Blob Storage Source

2.6.10

Azure Blob Storage Sink

1.6.27

Azure Data Lake Storage Gen2 Sink

1.6.27

Azure Cognitive Search Sink

1.1.7

Kafka Connect for Azure Cosmos DB (Source and Sink)

1.17.0

Azure Functions Sink

2.0.4

Azure Synapse Analytics Sink

1.0.9

Google BigQuery Sink

2.5.7

Cassandra Sink

2.0.10

Data Diode (Source and Sink)

1.2.6

Databricks Delta Lake Sink for AWS

1.0.19

Datadog Logs Sink

1.3.0

Debezium connector for MySQL

2.4.2

Debezium connector for PostgreSQL

2.5.4

Debezium connector for SQL Server

2.5.4

Elasticsearch Sink

14.1.2

Google Cloud Functions Sink

1.2.4

Google Cloud Storage Sink

10.2.1

Google Cloud Storage Source

10.2.1

GitHub Source

2.1.8

Google Cloud Pub/Sub Source

1.2.9

Google Cloud Spanner Sink

1.0.16

Google Firebase Realtime Database Connector (Source and Sink)

1.2.6

HDFS 3 Source

2.6.10

HDFS 3 Sink

1.2.4

HEAVY-AI (formerly OmniSci) Sink

1.0.9

HTTP Sink

1.7.8

HTTP Source

0.2.5

IBM MQ Sink

2.1.15

IBM MQ Source

12.2.9

InfluxDB Source

1.2.11

JDBC (Source and Sink)

10.8.2

Jira Source

1.2.13

JMS Sink Connector

2.1.15

JMS Source Connector

12.2.9

MongoDB Atlas Sink

1.15.0

Netezza Sink Connector

1.0.7

Oracle XStream CDC Source

1.1.0

PagerDuty Sink [Deprecated]

1.0.10

Redis Sink

0.0.8

Salesforce Bulk API (Source and Sink)

2.0.25

Salesforce (Source and Sink)

2.0.25

ServiceNow (Source and Sink)

2.5.4

SFTP (Source and Sink)

3.2.11

SNMP Trap Source

1.3.2

Snowflake Sink

3.1.1

Solace Source

1.2.8

Solace Sink

2.1.15

Splunk S2S Source

2.2.1

Splunk Source Connector

1.1.5

Kafka Connect Spooldir

2.0.67

Syslog Source

1.5.10

TIBCO EMS Sink

2.1.15

TIBCO EMS Source

1.2.9

Vertica Sink

1.3.2

VMware Tanzu Gemfire Sink

1.0.18

Zendesk Source Connector

1.3.4

Requirements

To use CSFLE or CSPE in Confluent Cloud with self-managed connectors, you must meet the following requirements:

  • Your Kafka cluster and Schema Registry must both be provisioned and running in Confluent Cloud.

Manage client-side encryption

At a high level, you can manage client-side encryption for connectors using the following 2-step process:

  1. Configure CSFLE or Configure CSPE

  2. Enable CSFLE or CSPE for connectors

Configure CSFLE

You can choose between the following two methods:

Configure CSFLE with shared KEK in Confluent

You must configure CSFLE in Confluent Cloud before you modify an existing connector or create a new one with CSFLE enabled. To configure CSFLE using KEK, follow the steps below:

  • Define the schema for the topic and add tags to the fields in the schema that you want to encrypt.

  • Create encryption keys for each KMS and allow Confluent Cloud to access your KMS.

  • Add encryption rules that specify the encryption key you want to use to encrypt the tags.

  • Grant DeveloperWrite permission for encryption key and DeveloperRead permission for the Schema Registry API keys.

For more information, see CSFLE with shared Confluent access to Key Encryption Keys (KEKs) .

Configure CSFLE without sharing KEK

If you do not want share your Key Encryption Key (KEK) with Confluent, follow the steps below:

  • Define the schema for the topic and add tags to the fields in the schema that you want to encrypt.

  • Create encryption keys for each KMS.

  • Add encryption rules that specify the encryption key you want to use to encrypt the tags.

  • Grant DeveloperWrite permission for encryption key and DeveloperRead permission for the Schema Registry API keys.

  • Add the following parameters in the connector configuration:

    For AWS, pass the following configuration parameters:

    Parameter

    Description

    rule.executors._default_.param.access.key.id=?

    The AWS access key identifier.

    rule.executors._default_.param.secret.access.key=?

    The AWS secret access key.

    For Azure, pass the following configuration parameters:

    Parameter

    Description

    rule.executors._default_.param.tenant.id

    The Azure tenant identifier.

    rule.executors._default_.param.client.id

    The Azure client identifier.

    rule.executors._default_.param.client.secret

    The Azure client secret.

    For Google Cloud, pass the following configuration parameters:

    Parameter

    Description

    rule.executors._default_.param.account.type

    This parameter contains the Google Cloud account type.

    rule.executors._default_.param.client.id

    The Google Cloud client identifier.

    rule.executors._default_.param.client.email

    The Google Cloud client email address.

    rule.executors._default_.param.private.key.id

    The Google Cloud private key identifier.

    rule.executors._default_.param.private.key

    The Google Cloud private key.

    For HashiCorp Vault, pass the following configuration parameters:

    Parameter

    Description

    rule.executors._default_.param.token.id

    The token identifier for HashiCorp Vault.

    rule.executors._default_.param.namespace

    The namespace for HashiCorp Vault Enterprise (optional).

For more information, see CSFLE without sharing access to your Key Encryption Keys (KEKs) .

Configure CSPE

You can choose between the following two methods:

Configure CSPE with shared KEK in Confluent

You must configure CSPE in Confluent Cloud before you modify an existing connector or create a new one with CSPE enabled. To configure CSPE using KEK, follow the steps below:

  • Define the schema for the topic that you want to encrypt.

  • Create encryption keys for each KMS and allow Confluent Cloud to access your KMS.

  • Add encoding rules that specify the encryption key you want to use for encryption.

  • Grant DeveloperWrite permission for encryption key and DeveloperRead permission for the Schema Registry API keys.

For more information, see CSPE with shared Confluent access to Key Encryption Keys (KEKs).

Configure CSPE without sharing KEK

If you do not want share your Key Encryption Key (KEK) with Confluent, follow the steps below:

  • Define the schema for the topic that you want to encrypt.

  • Create encryption keys for each KMS.

  • Add encoding rules that specify the encryption key you want to use for encryption.

  • Grant DeveloperWrite permission for encryption key and DeveloperRead permission for the Schema Registry API keys.

  • Add the following parameters in the connector configuration:

    For AWS, pass the following configuration parameters:

    Parameter

    Description

    rule.executors._default_.param.access.key.id=?

    The AWS access key identifier.

    rule.executors._default_.param.secret.access.key=?

    The AWS secret access key.

    For Azure, pass the following configuration parameters:

    Parameter

    Description

    rule.executors._default_.param.tenant.id

    The Azure tenant identifier.

    rule.executors._default_.param.client.id

    The Azure client identifier.

    rule.executors._default_.param.client.secret

    The Azure client secret.

    For Google Cloud, pass the following configuration parameters:

    Parameter

    Description

    rule.executors._default_.param.account.type

    This parameter contains the Google Cloud account type.

    rule.executors._default_.param.client.id

    The Google Cloud client identifier.

    rule.executors._default_.param.client.email

    The Google Cloud client email address.

    rule.executors._default_.param.private.key.id

    The Google Cloud private key identifier.

    rule.executors._default_.param.private.key

    The Google Cloud private key.

    For HashiCorp Vault, pass the following configuration parameters:

    Parameter

    Description

    rule.executors._default_.param.token.id

    The token identifier for HashiCorp Vault.

    rule.executors._default_.param.namespace

    The namespace for HashiCorp Vault Enterprise (optional).

For more information, see CSFLE without sharing access to your Key Encryption Keys (KEKs) .

Enable CSFLE or CSPE for connectors

To enable CSFLE or CSPE for connectors, define the following parameters with the mentioned boolean values in the connector configuration:

Note

If you do not add these values in the connector configuration, CSFLE or CSPE might not work properly.

  • csfle.enabled=true

  • value.converter.auto.register.schemas=false

  • value.converter.use.latest.version=true

  • key.converter.auto.register.schemas=false

  • key.converter.use.latest.version=true

Note

  • To fetch the latest value schema from Schema Registry, use value.converter.latest.cache.ttl.sec, that allows you to define the time interval, in seconds, after which the connector fetches the latest version of the value schema. By default, its value is set to -1. To enable it, enter the desired time interval in seconds for this parameter.

  • Similar to the value schema, use key.converter.latest.cache.ttl.sec to define the time interval, in seconds, after which the converter fetches the latest key schema from Schema Registry. The default value is -1. Change this value to the desired time interval in seconds.