Support newly joined consumers per issue#17

Author: omarsmak

src/main/kotlin/com/omarsmak/kafka/consumer/lag/monitoring/cli/ClientCli.kt

@@ -77,12 +77,9 @@ class ClientCli : Callable<Void> {
         return null
     }
 
-    private fun initializeConsumerGroups(): Set<String> {
-        val configConsumerGroups = kafkaConsumerClients.split(",")
-        return Utils.getTargetConsumerGroups(kafkaConsumerLagClient, configConsumerGroups)
-    }
+    private fun initializeConsumerGroups(): List<String> = kafkaConsumerClients.split(",")
 
-    private fun initializeConfigurations(targetConsumerGroups: Set<String>) = KafkaConsumerLagClientConfig.create(mapOf(
+    private fun initializeConfigurations(targetConsumerGroups: List<String>) = KafkaConsumerLagClientConfig.create(mapOf(
             KafkaConsumerLagClientConfig.BOOTSTRAP_SERVERS to kafkaBootstrapServers,
             KafkaConsumerLagClientConfig.HTTP_PORT to httpPort,
             KafkaConsumerLagClientConfig.POLL_INTERVAL to pollInterval,

src/main/kotlin/com/omarsmak/kafka/consumer/lag/monitoring/config/KafkaConsumerLagClientConfig.kt

@@ -28,7 +28,7 @@ open class KafkaConsumerLagClientConfig private constructor(
         private val httpPort by optional(name = HTTP_PORT, default = DEFAULT_HTTP_PORT)
         private val bootstrapServers by required<String>(name = BOOTSTRAP_SERVERS)
         private val pollInterval by optional(name = POLL_INTERVAL, default = DEFAULT_POLL_INTERVAL)
-        private val consumerGroups by required<Set<String>>(name = CONSUMER_GROUPS)
+        private val consumerGroups by required<List<String>>(name = CONSUMER_GROUPS)
         private val lagTreshold by optional(name = LAG_THRESHOLD, default = DEFAULT_LAG_THRESHOLD)
 
         /**

src/main/kotlin/com/omarsmak/kafka/consumer/lag/monitoring/response/ConsoleResponseView.kt

@@ -3,6 +3,7 @@
 package com.omarsmak.kafka.consumer.lag.monitoring.response
 
 import com.github.ajalt.mordant.TermColors
+import com.omarsmak.kafka.consumer.lag.monitoring.cli.Utils
 import com.omarsmak.kafka.consumer.lag.monitoring.client.KafkaConsumerLagClient
 import com.omarsmak.kafka.consumer.lag.monitoring.client.data.Lag
 import com.omarsmak.kafka.consumer.lag.monitoring.client.exceptions.KafkaConsumerLagClientException
@@ -24,7 +25,7 @@ class ConsoleResponseView : ResponseView {
     }
 
     override fun execute() {
-        val targetConsumerGroups: Set<String> = kafkaConsumerLagClientConfig[KafkaConsumerLagClientConfig.CONSUMER_GROUPS]
+        val targetConsumerGroups: List<String> = kafkaConsumerLagClientConfig[KafkaConsumerLagClientConfig.CONSUMER_GROUPS]
         val monitoringPollInterval: Long = kafkaConsumerLagClientConfig[KafkaConsumerLagClientConfig.POLL_INTERVAL]
         val monitoringLagThreshold: Int = kafkaConsumerLagClientConfig[KafkaConsumerLagClientConfig.LAG_THRESHOLD]
 
@@ -40,11 +41,13 @@ class ConsoleResponseView : ResponseView {
      * `Total consumer offsets: @totalConsumerOffsets`
      * `Total lag: @totalLag`
      */
-    private fun show(targetConsumerGroups: Set<String>, monitoringPollInterval: Long, monitoringLagThreshold: Int) {
+    private fun show(initialConsumerGroups: List<String>, monitoringPollInterval: Long, monitoringLagThreshold: Int) {
         Timer().scheduleAtFixedRate(0, monitoringPollInterval) {
             print("\u001b[H\u001b[2J")
 
-            targetConsumerGroups.forEach { consumer ->
+            val consumers = Utils.getTargetConsumerGroups(kafkaConsumerLagClient, initialConsumerGroups)
+
+            consumers.forEach { consumer ->
                 try {
                     val metrics = kafkaConsumerLagClient.getConsumerLag(consumer)
                     println("Consumer group: $consumer")

src/main/kotlin/com/omarsmak/kafka/consumer/lag/monitoring/response/PrometheusResponseView.kt

@@ -2,6 +2,7 @@
 
 package com.omarsmak.kafka.consumer.lag.monitoring.response
 
+import com.omarsmak.kafka.consumer.lag.monitoring.cli.Utils
 import com.omarsmak.kafka.consumer.lag.monitoring.client.KafkaConsumerLagClient
 import com.omarsmak.kafka.consumer.lag.monitoring.client.exceptions.KafkaConsumerLagClientException
 import com.omarsmak.kafka.consumer.lag.monitoring.config.KafkaConsumerLagClientConfig
@@ -59,7 +60,7 @@ class PrometheusResponseView : ResponseView {
     }
 
     override fun execute() {
-        val targetConsumerGroups: Set<String> = kafkaConsumerLagClientConfig[KafkaConsumerLagClientConfig.CONSUMER_GROUPS]
+        val targetConsumerGroups: List<String> = kafkaConsumerLagClientConfig[KafkaConsumerLagClientConfig.CONSUMER_GROUPS]
         val monitoringPollInterval: Long = kafkaConsumerLagClientConfig[KafkaConsumerLagClientConfig.POLL_INTERVAL]
         val httpPort: Int = kafkaConsumerLagClientConfig[KafkaConsumerLagClientConfig.HTTP_PORT]
 
@@ -75,15 +76,18 @@ class PrometheusResponseView : ResponseView {
      * `kafka_consumer_group_total_lag{group, topic}`
      * `kafka_topic_latest_offsets{group, topic, partition}
      */
-    private fun initialize(client: KafkaConsumerLagClient, targetConsumerGroups: Set<String>, port: Int, monitoringPollInterval: Long) {
+    private fun initialize(client: KafkaConsumerLagClient, initialConsumerGroups: List<String>, port: Int, monitoringPollInterval: Long) {
         // Start HTTP our server
         startServer(port)
 
         logger.info("Updating metrics every $monitoringPollInterval...")
 
         // Start publishing our metrics
         Timer().scheduleAtFixedRate(0, monitoringPollInterval) {
-            targetConsumerGroups.forEach { consumer ->
+
+            val consumers = Utils.getTargetConsumerGroups(kafkaConsumerLagClient, initialConsumerGroups)
+
+            consumers.forEach { consumer ->
                 try {
                     val lag = client.getConsumerLag(consumer)
 

src/test/kotlin/com/omarsmak/kafka/consumer/lag/monitoring/config/KafkaConsumerLagClientConfigTest.kt

@@ -17,7 +17,7 @@ internal class KafkaConsumerLagClientConfigTest {
         ))
 
         assertEquals("kafka1:9092,kafka2:9092", config[KafkaConsumerLagClientConfig.BOOTSTRAP_SERVERS])
-        assertEquals(setOf("consumer_1", "consumer_2"), config[KafkaConsumerLagClientConfig.CONSUMER_GROUPS])
+        assertEquals(listOf("consumer_1", "consumer_2"), config[KafkaConsumerLagClientConfig.CONSUMER_GROUPS])
         assertEquals(9000, config[KafkaConsumerLagClientConfig.HTTP_PORT])
         assertEquals(2000, config[KafkaConsumerLagClientConfig.POLL_INTERVAL])
         assertEquals(500, config[KafkaConsumerLagClientConfig.LAG_THRESHOLD])
@@ -38,7 +38,7 @@ internal class KafkaConsumerLagClientConfigTest {
         ))
 
         assertEquals("kafka1:9092,kafka2:9092", config[KafkaConsumerLagClientConfig.BOOTSTRAP_SERVERS])
-        assertEquals(setOf("consumer_1", "consumer_2"), config[KafkaConsumerLagClientConfig.CONSUMER_GROUPS])
+        assertEquals(listOf("consumer_1", "consumer_2"), config[KafkaConsumerLagClientConfig.CONSUMER_GROUPS])
         assertEquals(3000, config[KafkaConsumerLagClientConfig.HTTP_PORT])
         assertEquals(100, config[KafkaConsumerLagClientConfig.POLL_INTERVAL])
     }
@@ -84,7 +84,7 @@ internal class KafkaConsumerLagClientConfigTest {
         })
 
         assertEquals("kafka1:9092,kafka2:9092", config[KafkaConsumerLagClientConfig.BOOTSTRAP_SERVERS])
-        assertEquals(setOf("consumer_1", "consumer_2"), config[KafkaConsumerLagClientConfig.CONSUMER_GROUPS])
+        assertEquals(listOf("consumer_1", "consumer_2"), config[KafkaConsumerLagClientConfig.CONSUMER_GROUPS])
     }
 
     @Test