Akka Streams metrics

If you have an application written using Akka Streams and you are also using the Reactive Kafka connector for Akka Streams, then the topology client can publish metrics. The process is more complex than the one line change required for Kafka Streams but is still relatively simple.

First, you will require the following maven module:

<dependency>
    <groupId>com.landoop</groupId>
    <artifactId>lenses-topology-client-akka-streams-kafka_2.12</artifactId>
    <version>4.0.0</version>
</dependency>

For metrics on a producer, you must create a KafkaProducer from the usual producer settings before it is passed to the Akka stream as a sink and then register that producer with the topology client.

// setup your producer settings as normal
final ProducerSettings<String, String> producerSettings =
        ProducerSettings
                .create(system, 
                        new StringSerializer(), 
                        new StringSerializer())
                .withBootstrapServers("localhost:9092");

// using the producer settings, create a Kafka producer
KafkaProducer<String, String> producer = 
    producerSettings.createKafkaProducer();

// register the producer with the topology client along with the topic name you will publish to
client.register(
        topology.getAppName(), 
        outputTopic, 
        new KafkaMetricsBuilder(producer));

// create your sink as normal, except pass in the Kafka producer you created earlier
sink = Producer.plainSink(producerSettings, producer);

For consumer metrics, Reactive Kafka exposes these on the Consumer.Control instance that is returned as the materialized value. After the stream is materialized, you will have a handle to this control which can be registered directly with the topology.

// create your stream as normal, and retain a reference to the materialized value
Consumer.Control control = Consumer.plainSource(consumerSettings, Subscriptions.topics(inputTopic))
                            ...<stream-flow>...
                            .toMat(producer, Keep.left())
                            .run(materializer);

// register the control with the topology client
client.register(topology.getAppName(), inputTopic, new AkkaStreamsKafkaMetricBuilder(control));