Kafka Streams

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.

Installation

The Kpow Stream Agent can be found on Maven Central.

Include the agent as a dependency in your Kafka Streams application.

<dependency>
  <groupId>io.factorhouse</groupId>
  <artifactId>kpow-streams-agent</artifactId>
  <version>1.0.0</version>
  <type>bundle</type>
</dependency>

Documentation can be found at javadoc.io.

Integration

In your application, just before you start your KafkaStreams instance:

• Create a new io.factorhouse.kpow.StreamsRegistry instance.
• Register your KafkaStreams and Topology instances with the StreamsRegistry.

import io.factorhouse.kpow.StreamsRegistry;
import io.factorhouse.kpow.key.ClusterIdKeyStrategy;

// Your Kafka Streams topology
Topology topology = createMyTopology();

// Your Kafka Streams config
Properties props = new createMyStreamProperties();

// Your Kafka Streams instance
KafkaStreams streams = new KafkaStreams(topology, props);

// Create a Kpow StreamsRegistry
StreamsRegistry registry = new StreamsRegistry(props);

// Specify the key strategy when writing metrics to the internal Kafka topic
KeyStrategy keyStrategy = new ClusterIdKeyStrategy(props);

// Register your KafkaStreams and Topology instances with the StreamsRegistry
registry.register(streams, topology, keyStrategy);

// Start your Kafka Streams application
streams.start();

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.

Metric filters

You can configure each streams registry with metric filters, which give you greater control over which metrics Kpow's streams agent will export.

Metric filters can be chained and added programmatically:

import io.factorhouse.kpow.StreamsRegistry;
import io.factorhouse.kpow.MetricFilter;

MetricFilter metricFilter = MetricFilter().deny(); // don't send any streams metrics, just send through the Streams Topology

// ..

StreamsRegistry registry = new StreamsRegistry(props, metricFilter);

If you pass no metric filters to the StreamsRegistry constructor then the default metric filter will be used. The default metric filter will accept all metrics to be exported.

Metric filter usage

Kpow's streams agent metric filters work very similar to Micrometer's meter filters.

Metric filters can either ACCEPT or DENY a metric. The filter itself is a Java predicate which takes in the org.apache.common.MetricName class. This allows you to filter metrics by name, tags or group.

Metric filters are applied sequentially in the order they are configured in the registry. This allows for stacking of deny and accept filters to create more complex rules:

MetricFilter metricFilter = MetricFilter().acceptNameStartsWith("rocksdb").deny();

The above example allows all rocksdb related metrics through and denies all other types of streams metrics.

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
• sasl.oauthbearer.jwks.endpoint.url
• sasl.oauthbearer.token.endpoint.url
• sasl.kerberos.service.name

For more details visit the Producer section of the Apache Kafka documentation.

Key strategy

The keying strategy for data sent from Kpow's streams agent to its internal Kafka topic is configurable. The key strategy plays an important role in enabling Kpow to align stream metrics with the UI accurately. There are many key strategies available depending on your organisation's deployment.

Cluster ID

The default key strategy uses the cluster ID, obtained via an AdminClient describeClusters call. This AdminClient is created once during registry initialization and then closed. If you prefer not to have the streams registry create an AdminClient—either because your Kafka variant does not provide a cluster ID or due to security considerations—you may select an alternative key strategy from the options below.

// Specify the key strategy when writing metrics to the internal Kafka topic
// props are java.util.Properties describing the Kafka Connection
KeyStrategy keyStrategy = new ClusterIDKeyStrategy(props);
// Register your KafkaStreams and Topology instances with the StreamsRegistry
registry.register(streams, topology, keyStrategy);

Environment name

If you have set a UI-friendly cluster name using the ENVIRONMENT_NAME environment variable in Kpow, you can use this environment name as the keying strategy for the streams agent.

// This sets a manual key of `Trade Book (Staging)`, the name of the clusters environment name in Kpow's UI.
KeyStrategy keyStrategy = new ManualKeyStrategy("Trade Book (Staging)");
registry.

        register(streams, topology, keyStrategy);

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.

./kafka-acls.sh \
  --bootstrap-server 127.0.0.1:9092 \
  --command-config client.conf \
  --add --allow-principal User:<your-producer-user> --operation Write --topic '__oprtr_snapshot_state'

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.

Properties streamsProps = new Properties();
KafkaStreams streams = new KafkaStreams(topology, streamsProps);

StreamsRegistry registry = new StreamsRegistry(streamsProps);
...

Multi-Cluster Kpow

Use a Properties with your primary cluster configuration to create your StreamsRegistry.

Properties streamsProps = createMyStreamProperties();
KafkaStreams streams = new KafkaStreams(topology, streamsProps);

Properties primaryProps = createMyPrimaryClusterProducerProperties();
StreamsRegistry registry = new StreamsRegistry(primaryProps);
...

See the Multi-cluster management guide for more information.

Register Multiple Kafka Streams Instances

You can register multiple Kafka Streams instances on the same StreamsRegistry.

KafkaStreams dedupeStreams = new KafkaStreams(dedupeTopology, dedupeProps);
KafkaStreams paymentStreams = new KafkaStreams(paymentTopology, paymentProps);
registry.register(paymentStreams, keyStrategy);
registry.register(dedupeStreams, keyStrategy);

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