Kafka Streams natively integrates with the Apache Kafka® security features and supports all of the
client-side security features in Kafka. Streams leverages the Java Producer and Consumer API.
To secure your Stream processing applications, configure the security settings in the corresponding Kafka producer
and consumer clients, and then specify the corresponding configuration settings in your Kafka Streams application.
Kafka supports cluster encryption and authentication, including a mix of authenticated and unauthenticated,
and encrypted and non-encrypted clients. Using security is optional.
Here a few relevant client-side security features:
- Encrypt data-in-transit between your applications and Kafka brokers
- You can enable the encryption of the client-server communication between your applications and the Kafka brokers.
For example, you can configure your applications to always use encryption when reading and writing data to and from
Kafka. This is critical when reading and writing data across security domains such as internal network, public
internet, and partner networks.
- Client authentication
- You can enable client authentication for connections from your application to Kafka brokers. For example, you can
define that only specific applications are allowed to connect to your Kafka cluster.
- Client authorization
- You can enable client authorization of read and write operations by your applications. For example, you can define
that only specific applications are allowed to read from a Kafka topic. You can also restrict write access to Kafka
topics to prevent data pollution or fraudulent activities.
For more information about the security features in Kafka, see Kafka Security and the
blog post Apache Kafka Security 101.
Required ACL setting for secure Kafka clusters
Kafka clusters can use ACLs to control access to resources (like the ability to create topics), and for such clusters each client,
including Kafka Streams, is required to authenticate as a particular user in order to be authorized with appropriate access.
In particular, when Kafka Streams applications are run against a secured Kafka cluster, the principal running the application must have
the ACL set so that the application has the permissions to create internal topics.
To avoid providing this permission to your application, you can create the
required internal topics manually. If the internal topics exist, Kafka Streams
won’t try to recreate them.
The internal repartition and changelog topics must be created with the
correct number of partitions, otherwise Kafka Streams fails on startup.
The topics must be created with the same number of partitions as your
input topic. If there are multiple topics, they must be created with the
maximum number of partitions across all input topics.
Additionally, changelog topics must be created with log compaction
enabled, otherwise your application might lose data. For changelog topics
for windowed KTables, apply “delete,compact” and set the retention time
based on the corresponding store retention time. To avoid pre-mature
deletion, add a delta to the store retention time. By default, Kafka Streams
adds 24 hours to the store retention time.
You can find out more about the names of the required internal topics by using
Topology#describe(). All internal topics follow the naming pattern
<application.id>-<operatorName>-<suffix>, where the
suffix is either
There is no guarantee that this naming pattern will continue in future
releases, because it’s not part of the public API.
For best security, set only the minimum ACL operations. Allow only the
following operations for the Kafka Streams principal.
- Topic resource (for internal topics): READ, DELETE, WRITE, CREATE
- Consumer Group resource: READ, DESCRIBE
- Topic resource (for input topic): READ
- Topic resource (for output topic): WRITE
For example, given the following setup of your Kafka Streams application:
application.id value is
- Authenticating with the Kafka cluster as a
- The application’s coded topology reads from input topics
- The application’s topology write to output topics
The following commands create the necessary ACLs in the Kafka cluster to allow
your application to operate:
# Allow Streams to read the input topics:
bin/kafka-acls ... --add --allow-principal User:team1 --operation READ --topic input-topic1 --topic input-topic2
# Allow Streams to write to the output topics:
bin/kafka-acls ... --add --allow-principal User:team1 --operation WRITE --topic output-topic1 --topic output-topic2
# Allow Streams to manage its own internal topics:
bin/kafka-acls ... --add --allow-principal User:team1 --operation READ --operation DELETE --operation WRITE --operation CREATE --resource-pattern-type prefixed --topic team1-streams-app1
# Allow Streams to manage its own consumer groups:
bin/kafka-acls ... --add --allow-principal User:team1 --operation READ --operation DESCRIBE --resource-pattern-type prefixed --group team1-streams-app1
If you are planning to repartition topics in Kafka Streams, then be sure to
cleanup.policy=delete and also allow DELETE operations. The DELETE
operation ensures that, after repartitioning, the cleanup removes old records
from the logs. If you do not allow DELETE operations, then
there is a likelihood of increased file descriptor usage.
This example is based on the Confluent blog post
Apache Kafka Security 101.
The purpose is to configure a Kafka Streams application to enable client authentication and encrypt data-in-transit when
communicating with its Kafka cluster.
This example assumes that the Kafka brokers in the cluster already have their security setup and that the necessary SSL
certificates are available to the application in the local filesystem locations. For example, if you are using Docker
then you must also include these SSL certificates in the correct locations within the Docker image.
The snippet below shows the settings to enable client authentication and SSL encryption for data-in-transit between your
Kafka Streams application and the Kafka cluster it is reading and writing from:
# Essential security settings to enable client authentication and SSL encryption
Configure these settings in the application for your
StreamsConfig instance. These settings will encrypt any
data-in-transit that is being read from or written to Kafka, and your application will authenticate itself against the
Kafka brokers that it is communicating with. Note that this example does not cover client authorization.
// Code of your Java application that uses the Kafka Streams library
Properties settings = new Properties();
// Where to find secure Kafka brokers. Here, it's on port 9093.
// ...further non-security related settings may follow here...
// Security settings.
// 1. These settings must match the security settings of the secure Kafka cluster.
// 2. The SSL trust store and key store files must be locally accessible to the application.
If you incorrectly configure a security setting in your application, it will fail at runtime, typically right after you
start it. For example, if you enter an incorrect password for the
ssl.keystore.password setting, an error message
similar to this would be logged and then the application would terminate:
# Misconfigured ssl.keystore.password
Exception in thread "main" org.apache.kafka.common.KafkaException: Failed to construct kafka producer
Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException:
java.io.IOException: Keystore was tampered with, or password was incorrect
Caused by: java.security.UnrecoverableKeyException: Password verification failed
Monitor your Kafka Streams application log files for such error messages to spot any misconfigured applications quickly.