Kafka Streams

The Ultimate Engineering Toolkit for Apache Kafka®
Search…
Kafka Streams
Monitor Kafka Streams applications with kPow
Configure the kPow Streams Agent to unlock the following features:
Visualise Kafka Streams topologies in the kPow Streams UI.
Monitor Kafka Streams metrics (e.g Stream-Thread, State Store, RocksDB, etc).
See summaries of Kafka Streams activity for your Kafka cluster(s).
Aggregate and Expose Kafka Streams metrics via kPow Prometheus Endpoints (for alerting, etc).
(Soon) View kPow Insights of your Kafka Streams applications (outlier metrics, etc).
See the kPow Kafka Streams Spring Word Count Example for an integration of Kafka, Spring, and kPow.
The kPow Streams UI
Installation
The kPow Stream Agent can be found on Maven Central.
Include the agent as a dependency in your Kafka Streams application.
1
<dependency>
2
  <groupId>io.operatr</groupId>
3
  <artifactId>kpow-streams-agent</artifactId>
4
  <version>0.2.6</version>
5
  <type>bundle</type>
6
</dependency>
Copied!
Integration
In your application, just before you start your KafkaStreams instance:
Create a new io.operatr.kpow.StreamsRegistry instance.
Register your KafkaStreams and Topology instances with the StreamsRegistry.
1
import io.operatr.kpow.StreamsRegistry;
2
​
3
// Your Kafka Streams topology
4
Topology topology = createMyTopology(); 
5
​
6
// Your Kafka Streams config
7
Properties props = new createMyStreamProperties();
8
 
9
// Your Kafka Streams instance
10
KafkaStreams streams = new KafkaStreams(topology, props); 
11
​
12
// Create a kPow StreamsRegistry
13
StreamsRegistry registry = new StreamsRegistry(props);
14
​
15
// Register your KafkaStreams and Topology instances with the StreamsRegistry
16
registry.register(streams, topology); 
17
​
18
// Start your Kafka Streams application
19
streams.start();
Copied!
The StreamsRegistry is a single-threaded process that performs these actions once every minute:
Capture metadata about each registered Kafka Streams application.
Produce snapshots to the kPow internal __oprtr_snapshot_state topic.
The StreamsRegistry does not talk directly to kPow. kPow reads streams data from the snapshot topic.
Configuration
The StreamsRegistry Properties contains configuration to create the snapshot producer.
The StreamsRegistry configures its own Serdes on the snapshot producer, you do not need to set them.
Producer configuration means any of the following fields:
bootstrap.servers
ssl.truststore.type
ssl.truststore.password
ssl.truststore.location
ssl.truststore.certificates
ssl.trustmanager.algorithm
ssl.secure.random.implementation
ssl.provider
ssl.protocol
ssl.keystore.type
ssl.keystore.password
ssl.keystore.location
ssl.keystore.key
ssl.keystore.certificate.chain
ssl.keymanager.algorithm
ssl.key.password
ssl.endpoint.identification.algorithm
ssl.enabled.protocols
ssl.cipher.suites
security.protocol
sasl.mechanism
sasl.jaas.config
sasl.login.callback.handler.class
For more details visit the Producer section of the Apache Kafka documentation.
Minimum Required ACLs
If you secure your Kafka Cluster with ACLs, the user provided in the Producer configuration must have permission to write to the internal kPow topic.
1
./kafka-acls.sh \
2
  --bootstrap-server 127.0.0.1:9092 \
3
  --command-config client.conf \
4
  --add --allow-principal User:<your-producer-user> --operation Write --topic '__oprtr_snapshot_state'
Copied!
Produce to the Primary Cluster
When managing a single Kafka Cluster you can reuse the properties from your Kafka Streams application to create your StreamsRegisty. This is because the kPow internal topic ___oprtr_snapshot_compute lives in the cluster that your Kafka Streams application connects to.
When managing multiple Kafka Clusters configure your StreamsRegistry to produce snapshots to the primary Cluster that contains the internal kPow snapshot topics. This is the first cluster in your kPow configuration.
Single-Cluster kPow
Reuse your Kafka Streams Properties to create your StreamsRegistry.
1
Properties streamsProps = new Properties();
2
KafkaStreams streams = new KafkaStreams(topology, streamsProps);
3
​
4
StreamsRegistry registry = new StreamsRegistry(streamsProps);
5
...
Copied!
Multi-Cluster kPow
Use a Properties with your primary cluster configuration to create your StreamsRegistry.
1
Properties streamsProps = createMyStreamProperties();
2
KafkaStreams streams = new KafkaStreams(topology, streamsProps); 
3
​
4
Properties primaryProps = createMyPrimaryClusterProducerProperties();
5
StreamsRegistry registry = new StreamsRegistry(primaryProps);
6
...
Copied!
See the kPow Multi-Cluster Feature Guide for more information.
Multi-Cluster Configuration Feedback Request!
Is the requirement to produce to the primary kPow cluster difficult for you?
Please let us know - we are considering the option of always writing to the same cluster as your Kafka Streams connects to and having kPow gather snapshots from each cluster.
Register Multiple Kafka Streams Instances
You can register multiple Kafka Streams instances on the same StreamsRegistry.
1
KafkaStreams dedupeStreams = new KafkaStreams(dedupeTopology, dedupeProps);
2
KafkaStreams paymentStreams = new KafkaStreams(paymentTopology, paymentProps);
3
registry.register(paymentStreams, paymentTopology);
4
registry.register(dedupeStreams, dedupeTopology);
Copied!
View Streams Topologies
Each Kafka Streams application configured with the kPow Streams Agent is visible in the kPow Streams UI, navigate to the 'Workflows' tab for full topology visualisation.
Prometheus Egress
If you have enabled kPow's Prometheus integration, all Kafka Streams metrics collected from your running agents will be available from the endpoint /streams/v1
Staged Mutations
Next - Features
Data Governance (Audit Log)
Last modified 8mo ago
Copy link
Contents
Installation
Integration
Configuration
Minimum Required ACLs
Produce to the Primary Cluster
Single-Cluster kPow
Multi-Cluster kPow
View Streams Topologies
Prometheus Egress