Protect Data in Motion with TLS Encryption in Confluent Platform

TLS encryption overview

By default, Confluent Platform clusters communicate in PLAINTEXT, meaning that all data is sent in plain text (unencrypted). To encrypt data in motion (or data in transit) between services and components in your Confluent Platform cluster, you should configure all Confluent Platform services and components to use TLS encryption. TLS encryption is supported for all Confluent Platform services and components, including Confluent Server brokers, Connect workers, Kafka clients, and REST Proxy.

For details on configuring TLS encryption for Confluent Server brokers and Confluent Platform services, see the following sections:

Confluent Platform supports Transport Layer Security (TLS) encryption based on OpenSSL, an open source cryptography toolkit that provides an implementation of the Transport Layer Security (TLS) and Secure Socket Layer (SSL) protocols. Secure Sockets Layer (SSL) was the predecessor of Transport Layer Security (TLS), and has been deprecated since June 2015. For TLS configurations, properties often refer to SSL.

Enabling TLS encryption might have a performance impact due to overhead of encrypting and decrypting data.

TLS uses private-key/certificate pairs, which are used during the TLS handshake process.

  • Each Confluent Server broker needs its own private-key/certificate pair, and the Kafka client uses the certificate to authenticate to the Confluent Server broker.
  • Each logical client needs a private-key/certificate pair if client authentication is enabled, and the Confluent Server broker uses the certificate to authenticate the Kafka client.

You can configure each Confluent Server broker and logical client with a truststore, which is used to determine which certificates (broker or logical client identities) to trust (authenticate). You can configure the truststore in many ways. Consider the following two examples:

  • The truststore contains one or many certificates: the Confluent Server broker or logical Kafka client trusts any certificate listed in the truststore.
  • The truststore contains a Certificate Authority (CA): the Confluent Server broker or logical Kafka client trusts any certificate that was signed by the CA in the truststore.

Using the CA method is more convenient because adding a new Confluent Server broker or Kafka client doesn’t require changes to the truststore. For an example of the CA method, see Single Trust Store Configuration.

However, with the CA method, Confluent Platform clusters do not conveniently support blocking authentication for individual Confluent Server brokers or Kafka clients that were previously trusted using this mechanism (certificate revocation is typically done using Certificate Revocation Lists or the Online Certificate Status Protocol), so you would have to rely on authorization to block access.

In contrast, if you use one or many certificates, blocking authentication is achieved by removing the Confluent Server broker certificate or Kafka client certificate from the truststore.

See also

For an example that shows how to set Docker environment variables for Confluent Platform running in ZooKeeper mode, see the Confluent Platform demo. Refer to the demo’s docker-compose.yml file for a configuration reference.

Mutual (mTLS) authentication

If you configure TLS encryption, you can optionally configure mutual (mTLS) authentication. You can configure just TLS encryption (by default, TLS encryption includes certificate authentication of the server) and use a separate mechanism for client authentication (for example, mTLS or SASL). By default, TLS encryption enables one-way authentication in which the client authenticates the server certificate. For bidirectional authentication, where the broker also authenticates the client certificate, you can use mTLS.

When you use mutual TLS (mTLS) authentication, the Confluent Server broker authenticates the Kafka client and the Kafka client also authenticates the Confluent Server broker. This bidirectional, or mutual, authentication provides an additional layer of security for your Confluent Platform cluster. Note that mTLS is not supported for Kafka clients authenticating to ksqlDB.

Create TLS keys and certificates

Refer to the Security Tutorial, which describes how to create TLS keys and certificates.

Configure TLS encryption for Confluent Server brokers

Configure all Confluent Server brokers in your Confluent Platform cluster to accept secure connections from clients. Any configuration changes made to brokers require a rolling restart.

Enable security for Confluent Server brokers as described in the section below. Additionally, if you are using Confluent Control Center, Auto Data Balancer, or Self-Balancing Clusters, configure your brokers for:

  1. Configure the password, truststore, and keystore in the server.properties file of every Confluent Server broker. Because this configuration stores passwords directly in the broker configuration file, restrict access to these files using file system permissions.

    ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks
ssl.truststore.password=test1234
ssl.keystore.location=/var/ssl/private/kafka.server.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234

  2. To enable TLS for interbroker communication, add the following to the Confluent Server broker properties file (default is PLAINTEXT):

    security.inter.broker.protocol=SSL

  3. Configure the Confluent Server brokers with the ports to listen for client and interbroker TLS (SSL) connections. If the value is different from listeners, you must configure listeners, and optionally advertised.listeners.

    listeners=SSL://kafka1:9093
advertised.listeners=SSL://<localhost>:9093

  4. Configure both SSL ports and PLAINTEXT ports if:

    • TLS is not enabled for interbroker communication
    • Some Kafka clients connecting to the cluster do not use TLS
    listeners=PLAINTEXT://kafka1:9092,SSL://kafka1:9093
advertised.listeners=PLAINTEXT://<localhost>:9092,SSL://<localhost>:9093

    Note that advertised.host.name and advertised.port configure a single PLAINTEXT port and are incompatible with secure protocols. Use advertised.listeners instead.

Optional settings

Here are some optional settings for Confluent Server brokers:

ssl.cipher.suites

The named list of supported cipher suites, which are named combinations of authentication, encryption, MAC, and key exchange algorithms used to negotiate the security settings for a network connection using TLS.

  • Type: list
  • Default: null (by default, all supported cipher suites are enabled)
  • Importance: medium
ssl.enabled.protocols

The comma-separated list of protocols enabled for TLS connections. Kafka clients and Confluent Server brokers use TLSv1.3 if both support it, but fall back to TLSv1.2 assuming both support TLSv1.2.

  • Type: list
  • Default: TLSv1.2,TLSv1.3
  • Importance: medium
ssl.truststore.type

The file format of the truststore file.

  • Type: string
  • Default: JKS
  • Importance: medium

Due to import regulations in some countries, the Oracle implementation limits the strength of cryptographic algorithms available by default. If stronger algorithms are needed (for example, AES with 256-bit keys), the JCE Unlimited Strength Jurisdiction Policy Files must be obtained and installed in the JDK or JRE. See JCA Providers Documentation for more information.

Configure TLS encryption for Kafka clients

The new Producer and Consumer clients support security for Kafka versions 0.9.0 and higher.

If you are using the Kafka Streams API, you can read on how to configure equivalent SSL and SASL parameters.

If client authentication is not required by the Confluent Server broker, the following is a minimal configuration example that you can store in a client properties file client-ssl.properties.

Because this configuration stores passwords directly in the Kafka client configuration file, it is important to restrict access to these files via file system permissions.

bootstrap.servers=kafka1:9093
security.protocol=SSL
ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks
ssl.truststore.password=test1234

If client authentication requires TLS, the client must provide the keystore as well. You can read about the additional configurations required in mTLS authentication.

Here are examples using Kafka tools, kafka-console-producer and kafka-console-consumer, to pass in the client-ssl.properties file with the properties specified above:

kafka-console-producer --broker-list kafka1:9093 \
  --topic test \
  --producer.config client-ssl.properties
kafka-console-consumer --bootstrap-server kafka1:9093 \
  --topic test \
  --consumer.config client-ssl.properties \
  --from-beginning

Optional settings

Here are some commonly used optional settings for Kafka clients:

ssl.provider

The name of the security provider used for TLS connections. Default value is the default security provider of the JVM.

  • Type: string
  • Default: null
  • Importance: medium
ssl.cipher.suites

The named list of supported cipher suites, which are named combinations of authentication, encryption, MAC, and key exchange algorithms used to negotiate the security settings for a network connection using TLS.

  • Type: list
  • Default: null (by default, all supported cipher suites are enabled)
  • Importance: medium
ssl.enabled.protocols

The comma-separated list of protocols enabled for TLS connections. The default value is TLSv1.2,TLSv1.3 when running with Java 11 or later, TLSv1.2 otherwise. With the default value for Java 11 (TLSv1.2,TLSv1.3), Kafka clients and brokers prefer TLSv1.3 if both support it, but fall back to TLSv1.2 (assuming both support at least TLSv1.2).

  • Type: list
  • Default: TLSv1.2,TLSv1.3
  • Importance: medium
ssl.truststore.type

The file format of the truststore file.

  • Type: string
  • Default: JKS
  • Importance: medium

Configure TLS encryption for ZooKeeper

The version of ZooKeeper bundled with Kafka supports TLS encryption. For details, see Add Security to Running Clusters in Confluent Platform.

Configure TLS encryption for Connect workers

This section describes how to enable security for Kafka Connect. Securing Kafka Connect requires that you configure security for:

  1. Kafka Connect workers: part of the Kafka Connect API, a worker is really just an advanced client, underneath the covers
  2. Kafka Connect connectors: connectors may have embedded producers or consumers, so you must override the default configurations for Connect producers used with source connectors and Connect consumers used with sink connectors
  3. Kafka Connect REST: Kafka Connect exposes a REST API that can be configured to use TLS/SSL using additional properties

Configure security for Kafka Connect as described in the section below. Additionally, if you are using Confluent Control Center streams monitoring for Kafka Connect, configure security for:

Configure the top-level settings in the Connect workers to use TLS by adding these properties in connect-distributed.properties. These top-level settings are used by the Connect worker for group coordination and to read and write to the internal topics which are used to track the cluster’s state (for example, configurations and offsets).

bootstrap.servers=kafka1:9093
security.protocol=SSL
ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks
ssl.truststore.password=test1234

Connect workers manage the producers used by source connectors and the consumers used by sink connectors. So, for the connectors to leverage security, you also have to override the default producer or consumer configuration that the worker uses. Depending on whether the connector is a source or sink connector:

  • For source connectors, configure the same properties, but add the producer prefix.

    producer.bootstrap.servers=kafka1:9093
producer.security.protocol=SSL
producer.ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks
producer.ssl.truststore.password=test1234

  • For sink connectors, configure the same properties, but add the consumer prefix.

    consumer.bootstrap.servers=kafka1:9093
consumer.security.protocol=SSL
consumer.ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks
consumer.ssl.truststore.password=test1234

Configure TLS encryption for REST Proxy

  • For TLS encryption with the REST API, configure the following additional properties:
Property Note
listeners
List of REST listeners in the format
protocol://host:port,protocol2://host2:port,
where the protocol is either http or https.
rest.advertised.listener
Configures the listener used for communication between workers.
Valid values are either http or https.
If the listeners property is not defined or if it contains an http
listener, the default value for this field is http. When the listeners
property is defined and contains only https listeners, the default value is
https.
ssl.client.auth
Valid values are none, requested, and required. It controls whether:
1. the client is required to do TLS client authentication (required)
2. it can decide to skip the TLS client authentication (requested)
3. the TLS client authentication will be disabled (none)
listeners.https.ssl.*
You can use the listeners.https. prefix with an TLS configuration parameter
to override the default TLS configuration that is shared with the connections
to the Confluent Server broker. If at least one parameter with this prefix exists, the
implementation uses only the TLS parameters with this prefix and ignores all TLS
parameters without this prefix. If no parameter with prefix listeners.https.
exists, the parameters without a prefix are used.

Note that if the listeners.https.ssl.* properties are not defined then the ssl.* properties will be used. For a list of all REST API ssl.* properties, see REST Proxy Configuration Options.

Here is an example that sets the ssl.* properties to use TLS connections to the Confluent Server broker, and since listeners includes https these same settings are used to configure the Connect TLS endpoint:

listeners=https://myhost:8443
rest.advertised.listener=https
rest.advertised.host.name=<localhost>
rest.advertised.host.port=8083
ssl.client.auth=requested
ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks
ssl.truststore.password=test1234
ssl.keystore.location=/var/ssl/private/kafka.server.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234

To configure the Connect TLS endpoint differently than the TLS connections to the Confluent Server broker, simply define the listeners.https.ssl.* properties with the correct settings. Note that as soon as any listeners.https.ssl.* properties are specified, none of the top level ssl.* properties will apply, so be sure to define all of the necessary listeners.https.ssl.* properties:

listeners=https://myhost:8443
rest.advertised.listener=https
rest.advertised.host.name=<localhost>
rest.advertised.host.port=8083
listeners.https.ssl.client.authentication=requested
listeners.https.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks
listeners.https.ssl.truststore.password=test1234
listeners.https.ssl.keystore.location=/var/ssl/private/kafka.server.keystore.jks
listeners.https.ssl.keystore.password=test1234
listeners.https.ssl.key.password=test1234
listeners.https.ssl.endpoint.identification.algorithm=HTTPS

Configure TLS encryption for Replicator

Confluent Replicator is a type of Kafka source connector that replicates data from a source to destination Kafka cluster. An embedded consumer inside Replicator consumes data from the source cluster, and an embedded producer inside the Kafka Connect worker produces data to the destination cluster.

Replicator version 4.0 and earlier requires a connection to ZooKeeper in the origin and destination Kafka clusters. If ZooKeeper is configured for authentication, the client configures the ZooKeeper security credentials via the global JAAS configuration setting -Djava.security.auth.login.config on the Connect workers, and the ZooKeeper security credentials in the origin and destination clusters must be the same.

To configure Confluent Replicator security, you must configure the Replicator connector as shown below and additionally you must configure:

To add TLS encryption to the Confluent Replicator embedded consumer, modify the Replicator JSON properties file.

Here is an example subset of configuration properties to add for TLS encryption:

{
  "name":"replicator",
    "config":{
      ....
      "src.kafka.ssl.truststore.location":"/etc/kafka/secrets/kafka.connect.truststore.jks",
      "src.kafka.ssl.truststore.password":"confluent",
      "src.kafka.security.protocol":"SSL"
      ....
    }
  }
}

See also

To see an example Confluent Replicator configuration, see the TLS source demo script. For demos of common security configurations see: Replicator security demos

To configure Confluent Replicator for a destination cluster with TLS encryption, modify the Replicator JSON configuration to include the following:

{
  "name":"replicator",
    "config":{
      ....
      "dest.kafka.ssl.truststore.location":"/etc/kafka/secrets/kafka.connect.truststore.jks",
      "dest.kafka.ssl.truststore.password":"confluent",
      "dest.kafka.security.protocol":"SSL"
      ....
    }
  }
}

Additionally the following properties are required in the Connect worker:

security.protocol=SSL
ssl.truststore.location=/etc/kafka/secrets/kafka.connect.truststore.jks
ssl.truststore.password=confluent
producer.security.protocol=SSL
producer.ssl.truststore.location=/etc/kafka/secrets/kafka.connect.truststore.jks
producer.ssl.truststore.password=confluent

For details, see the general security configuration for Connect workers.

See also

For an example Confluent Replicator configuration, see TLS destination demo script.

For demos of common security configurations, see: Replicator security demos.

Configure TLS encryption for Confluent Control Center

Confluent Control Center uses Kafka Streams as a state store, so if all the Kafka brokers in the cluster backing Control Center are secured, then the Control Center application also needs to be secured.

Note

When RBAC is enabled, Control Center cannot be used in conjunction with Kerberos because Control Center cannot support any SASL mechanism other than OAUTHBEARER.

Enable security for the Control Center application as described in the section below. Additionally, configure security for the following components:

Enable TLS encryption for Confluent Control Center in the etc/confluent-control-center/control-center.properties file.

confluent.controlcenter.streams.security.protocol=SSL
confluent.controlcenter.streams.ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks
confluent.controlcenter.streams.ssl.truststore.password=test1234

Configure TLS encryption for Metrics Reporter

This section describes how to enable TLS encryption for Confluent Metrics Reporter, which is used for Confluent Control Center and Auto Data Balancer.

To add TLS encryption for the Confluent Metrics Reporter, add the following to the server.properties file on the Confluent Server brokers in the Confluent Platform cluster being monitored.

confluent.metrics.reporter.bootstrap.servers=kafka1:9093
confluent.metrics.reporter.security.protocol=SSL
confluent.metrics.reporter.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks
confluent.metrics.reporter.ssl.truststore.password=test1234

Configure TLS encryption for Confluent Monitoring Interceptors

Confluent Monitoring Interceptors are used for Confluent Control Center streams monitoring. This section describes how to enable security for Confluent Monitoring Interceptors in three places:

  1. General clients
  2. Kafka Connect
  3. Confluent Replicator

Important

The typical use case for Confluent Monitoring Interceptors is to provide monitoring data to a separate monitoring cluster that most likely has different configurations. Interceptor configurations do not inherit configurations for the monitored component. If you wish to use configurations from the monitored component, you must add the appropriate prefix. For example, the option confluent.monitoring.interceptor.security.protocol=SSL, if being used for a producer, must be prefixed with producer. and would appear as producer.confluent.monitoring.interceptor.security.protocol=SSL.

Interceptors for general clients

For Confluent Control Center stream monitoring to work with Kafka clients, you must configure TLS encryption for the Confluent Monitoring Interceptors in each client.

  1. Verify that the client has configured interceptors.

  • Producer:

    interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor

  • Consumer:

    interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor

  1. Configure TLS encryption for the interceptor.

    confluent.monitoring.interceptor.bootstrap.servers=kafka1:9093
confluent.monitoring.interceptor.security.protocol=SSL
confluent.monitoring.interceptor.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks
confluent.monitoring.interceptor.ssl.truststore.password=test1234

Interceptors for Kafka Connect

For Confluent Control Center stream monitoring to work with Kafka Connect, you must configure TLS encryption for the Confluent Monitoring Interceptors in Kafka Connect. Configure the Connect workers by adding the following properties in connect-distributed.properties, depending on whether the connectors are sources or sinks.

  • Source connector: configure the Confluent Monitoring Interceptors for TLS encryption, adding producer prefix.

    producer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor
producer.confluent.monitoring.interceptor.bootstrap.servers=kafka1:9093
producer.confluent.monitoring.interceptor.security.protocol=SSL
producer.confluent.monitoring.interceptor.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks
producer.confluent.monitoring.interceptor.ssl.truststore.password=test1234

  • Sink connector: configure the Confluent Monitoring Interceptors for TLS encryption, adding the consumer prefix.

    consumer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor
consumer.confluent.monitoring.interceptor.bootstrap.servers=kafka1:9093
consumer.confluent.monitoring.interceptor.security.protocol=SSL
consumer.confluent.monitoring.interceptor.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks
consumer.confluent.monitoring.interceptor.ssl.truststore.password=test1234

Interceptors for Replicator

For Confluent Control Center stream monitoring to work with Replicator, you must configure TLS for the Confluent Monitoring Interceptors in the Replicator JSON configuration file. Here is an example subset of configuration properties to add for TLS encryption.

{
  "name":"replicator",
    "config":{
      ....
      "src.consumer.group.id": "replicator",
      "src.consumer.interceptor.classes": "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor",
      "src.consumer.confluent.monitoring.interceptor.bootstrap.servers": "kafka1:9093",
      "src.consumer.confluent.monitoring.interceptor.security.protocol": "SSL",
      "src.consumer.confluent.monitoring.interceptor.ssl.truststore.location": "/var/ssl/private/kafka.client.truststore.jks",
      "src.consumer.confluent.monitoring.interceptor.ssl.truststore.password": "test1234",
      ....
    }
  }
}

Enable TLS encryption in a Self-Balancing cluster

To enable TLS encryption in a Self-Balancing cluster, add the following to the server.properties file on the Confluent Server brokers in the Confluent Platform cluster.

confluent.rebalancer.metrics.security.protocol=SSL
confluent.rebalancer.metrics.ssl.truststore.location=/etc/kafka/secrets/kafka.client.truststore.jks
confluent.rebalancer.metrics.ssl.truststore.password=confluent
confluent.rebalancer.metrics.ssl.keystore.location=/etc/kafka/secrets/kafka.client.keystore.jks
confluent.rebalancer.metrics.ssl.keystore.password=confluent
confluent.rebalancer.metrics.ssl.key.password=confluent

Schema Registry

Schema Registry uses Kafka to persist schemas, and so it acts as a client to write data to the Kafka cluster. Therefore, if the Kafka brokers are configured for security, you should also configure Schema Registry to use security. You may also refer to the complete list of Schema Registry configuration options.

Here is an example subset of schema-registry.properties configuration parameters to add for TLS encryption:

kafkastore.bootstrap.servers=SSL://kafka1:9093
kafkastore.security.protocol=SSL
kafkastore.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks
kafkastore.ssl.truststore.password=test1234

Configure TLS encryption for REST Proxy

To use TLS encryption with Confluent REST Proxy, you need to configure security between:

  • REST clients and the REST Proxy (HTTPS)
  • REST Proxy and the Confluent Platform cluster

For the complete list of options, see REST Proxy configuration options.

First, configure HTTPS between REST clients and the REST Proxy. Here is an example subset of kafka-rest.properties configuration parameters to configure HTTPS:

ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks
ssl.truststore.password=test1234

Then, configure TLS encryption between REST Proxy and the Confluent Platform cluster. Here is an example subset of kafka-rest.properties configuration parameters to add for TLS encryption:

client.bootstrap.servers=kafka1:9093
client.security.protocol=SSL
client.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks
client.ssl.truststore.password=test1234

Enable TLS logging

Enable TLS debug logging at the JVM level by starting the Confluent Server broker and clients with the javax.net.debug system property. For example:

export KAFKA_OPTS=-Djavax.net.debug=all
kafka-server-start etc/kafka/server.properties

Tip

These instructions are based on the assumption that you are installing Confluent Platform by using ZIP or TAR archives. For more information, see Install Confluent Platform On-Premises.

Once you start the Confluent Server broker, you should see the following in the server.log:

with addresses: PLAINTEXT -> EndPoint(192.168.64.1,9092,PLAINTEXT),SSL -> EndPoint(192.168.64.1,9093,SSL)

To verify that the server keystore and truststore are configured correctly, run the following openssl s_client command:

openssl s_client -debug -connect localhost:9093 -tls1

Note: TLSv1 should be listed under ssl.enabled.protocols.

In the output of this command you should see the server certificate:

-----BEGIN CERTIFICATE-----
{variable sized random bytes}
-----END CERTIFICATE-----
subject=/C=US/ST=CA/L=Santa Clara/O=org/OU=org/CN=Joe Smith
issuer=/C=US/ST=CA/L=Santa Clara/O=org/OU=org/CN=kafka/emailAddress=test@test.com

For details, see Debugging SSL/TLS Connections [Java Documentation].

If the certificate is not returned using the openssl command, or if there are other error messages, then your keys or certificates are not set up correctly. Review your configurations.