Add consumer group view (#155)

* Add describe consumer group to adminApi

* Port view and view model from previous version

* Draft consumer group view

* Reorganize insulator.di package

* Style treeview

* Fix PR lint

* Add integration test for kafka conmsumer group

* Add integration test for the consumer group view

* Use dark theme in integration tests

* Fix PR lint

* Update Readme.md

* Update Readme.md

Co-authored-by: Auto Lint <[email protected]>

.github/workflows/ci.yml

@@ -85,7 +85,7 @@ jobs:
             ${{github.workspace}}/lib/kafka/src/main/kotlin \
             ${{github.workspace}}/lib/update/src/main/kotlin
         with:
-          coverageCommand: "xvfb-run --auto-servernum ./gradlew check :app:integrationTest codeCoverageReport --no-daemon"
+          coverageCommand: "xvfb-run --auto-servernum ./gradlew check integrationTest codeCoverageReport --no-daemon"
           coverageLocations: ${{github.workspace}}/build/reports/jacoco/codeCoverageReport/codeCoverageReport.xml:jacoco
       - name: Configure AWS credentials
         if: ${{ matrix.os.run-integration }}

.github/workflows/release.yml

@@ -1,4 +1,4 @@
-nrame: Release
+name: Release
 
 on:
   push:

Readme.md

@@ -82,6 +82,7 @@ Download the binary from the latest release for your OS. Learn more [here](https
   * String producer
 * 🚧  **Consumer groups** 🚧
   * List consumer groups
+  * Show topics, partitions and lags
 * **Cross platform**
   * Windows, macOS and Linux ready.
 * **Dark/Light theme**

app/src/integrationTest/kotlin/insulator/integrationtest/ConsumerGroupsTests.kt

@@ -2,18 +2,27 @@ package insulator.integrationtest
 
 import insulator.integrationtest.helpers.IntegrationTestFixture
 import insulator.integrationtest.helpers.click
+import insulator.integrationtest.helpers.doubleClick
 import insulator.integrationtest.helpers.lookupFirst
+import insulator.integrationtest.helpers.screenShoot
 import insulator.integrationtest.helpers.selectCluster
 import insulator.integrationtest.helpers.waitWindowWithTitle
+import insulator.kafka.factories.kafkaConfig
 import io.kotest.core.spec.style.FreeSpec
 import io.kotest.matchers.shouldBe
 import javafx.scene.Node
 import javafx.scene.control.Label
+import javafx.scene.control.TreeView
+import kotlinx.coroutines.delay
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.clients.consumer.KafkaConsumer
 import org.apache.kafka.common.serialization.StringDeserializer
 import tornadofx.CssRule
+import tornadofx.expandAll
+import java.io.Closeable
 import java.time.Duration
+import java.util.Properties
+import kotlin.concurrent.thread
 import kotlin.time.ExperimentalTime
 
 @ExperimentalTime
@@ -24,28 +33,19 @@ class ConsumerGroupsTests : FreeSpec({
             val clusterName = "Test cluster"
             val groupIdName = "test-consumer-group-id"
             val testTopic = "topic-consumer-topic"
+            val consumerNumber = 3
 
             fixture.startAppWithKafkaCuster(clusterName, false)
 
             // Create topic and send a message
             fixture.createTopic(testTopic)
             fixture.stringProducer.send(testTopic, "k", "v")
             // Create a consumer group
-            with(
-                KafkaConsumer<String, String>(
-                    mapOf(
-                        ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to fixture.currentKafkaCluster.endpoint,
-                        ConsumerConfig.GROUP_ID_CONFIG to groupIdName,
-                        ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
-                        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
-                    )
-                )
-            ) {
-                subscribe(listOf(testTopic))
-                poll(Duration.ofMillis(300))
-                commitSync()
-                close()
+            val consumerConfig = kafkaConfig(fixture.currentKafkaCluster).apply {
+                put(ConsumerConfig.GROUP_ID_CONFIG, groupIdName)
+                put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java)
             }
+            val consumers = (1..consumerNumber).map { testConsumer(consumerConfig, testTopic) }
 
             // open main view
             selectCluster(fixture.currentKafkaCluster)
@@ -54,6 +54,37 @@ class ConsumerGroupsTests : FreeSpec({
             mainView.lookupFirst<Node>(CssRule.id("sidebar-item-consumer-group")).click()
             // the consumer group is shown in the list of consumers
             mainView.lookupFirst<Label>(CssRule.id("consumer-$groupIdName")).text shouldBe groupIdName
+            delay(10_000) // wait a bit for re-balancing
+            // double click on the consumer group name shows the the tree view with the lag
+            mainView.lookupFirst<Label>(CssRule.id("consumer-$groupIdName")).doubleClick()
+            with(mainView.lookupFirst<TreeView<String>>(".tree-view")) {
+                root.value shouldBe "Consumers"
+                root.children.size shouldBe consumerNumber
+                root.expandAll()
+            }
+            screenShoot("consumer-group-view")
+            consumers.forEach { it.close() }
         }
     }
 })
+
+fun testConsumer(consumerConfig: Properties, topicName: String) = object : Closeable {
+
+    private var isRunning = true
+
+    private val consumerThread = with(KafkaConsumer<String, String>(consumerConfig)) {
+        subscribe(listOf(topicName))
+        thread(start = true) {
+            while (isRunning) {
+                poll(Duration.ofMillis(200))
+                commitSync()
+            }
+            this.close()
+        }
+    }
+
+    override fun close() {
+        isRunning = false
+        consumerThread.join()
+    }
+}

app/src/integrationTest/kotlin/insulator/integrationtest/helpers/IntegrationTestFixture.kt

@@ -2,6 +2,7 @@ package insulator.integrationtest.helpers
 
 import insulator.Insulator
 import insulator.configuration.ConfigurationRepo
+import insulator.configuration.model.InsulatorTheme
 import insulator.kafka.AdminApi
 import insulator.kafka.SchemaRegistry
 import insulator.kafka.adminApi
@@ -73,6 +74,7 @@ class IntegrationTestFixture : Closeable {
     private suspend fun storeConfiguration(vararg cluster: Cluster) =
         ConfigurationRepo("$currentHomeFolder/.insulator.config").let { repo ->
             cluster.forEach { repo.store(it) }
+            repo.store(InsulatorTheme.Dark)
         }
 
     override fun close() {

app/src/main/kotlin/insulator/App.kt

@@ -1,6 +1,6 @@
 package insulator
 
-import insulator.di.components.DaggerInsulatorComponent
+import insulator.di.DaggerInsulatorComponent
 import insulator.helper.dispatch
 import insulator.ui.style.AppBarStyle
 import insulator.ui.style.ButtonStyle
@@ -14,6 +14,7 @@ import insulator.ui.style.ScrollBarStyle
 import insulator.ui.style.ScrollPaneStyle
 import insulator.ui.style.TableViewStyle
 import insulator.ui.style.TextStyle
+import insulator.ui.style.TreeViewStyle
 import javafx.stage.Stage
 import tornadofx.App
 import tornadofx.FX
@@ -33,7 +34,8 @@ class Insulator : App(
     DialogPaneStyle::class,
     ScrollPaneStyle::class,
     MainViewStyle::class,
-    ScrollBarStyle::class
+    ScrollBarStyle::class,
+    TreeViewStyle::class
 ) {
     private val daggerInsulator = DaggerInsulatorComponent.builder().build()
 

app/src/main/kotlin/insulator/di/components/ClusterComponent.kt → app/src/main/kotlin/insulator/di/ClusterComponent.kt

@@ -1,8 +1,8 @@
-package insulator.di.components
+package insulator.di
 
 import dagger.BindsInstance
 import dagger.Component
-import insulator.di.ClusterScope
+import insulator.CachedFactory
 import insulator.di.modules.ClusterModule
 import insulator.jsonhelper.JsonFormatter
 import insulator.jsonhelper.avrotojson.AvroToJsonConverter
@@ -17,6 +17,16 @@ import insulator.views.main.MainView
 import insulator.views.main.consumergroup.ListConsumerGroupView
 import insulator.views.main.schemaregistry.ListSchemaView
 import insulator.views.main.topic.ListTopicView
+import javax.inject.Inject
+import javax.inject.Scope
+
+@Scope
+annotation class ClusterScope
+
+class ClusterComponentFactory @Inject constructor(insulatorComponent: InsulatorComponent) :
+    CachedFactory<Cluster, ClusterComponent>({ cluster ->
+        DaggerClusterComponent.factory().build(insulatorComponent, cluster)
+    })
 
 @ClusterScope
 @Component(dependencies = [InsulatorComponent::class], modules = [ClusterModule::class])

app/src/main/kotlin/insulator/di/ConsumerGroupComponent.kt

@@ -0,0 +1,32 @@
+package insulator.di
+
+import dagger.BindsInstance
+import dagger.Component
+import insulator.CachedFactory
+import insulator.kafka.model.Cluster
+import insulator.views.main.consumergroup.ConsumerGroupView
+import javax.inject.Inject
+import javax.inject.Scope
+
+@Scope
+annotation class ConsumerGroupScope
+data class ConsumerGroupId(val id: String)
+
+class ConsumerGroupComponentFactory @Inject constructor(clusterComponent: ClusterComponent) :
+    CachedFactory<ConsumerGroupId, ConsumerGroupComponent>({ consumerGroup: ConsumerGroupId ->
+        DaggerConsumerGroupComponent.factory().build(clusterComponent, consumerGroup)
+    })
+
+@ConsumerGroupScope
+@Component(dependencies = [ClusterComponent::class])
+interface ConsumerGroupComponent {
+
+    @Component.Factory
+    interface Factory {
+        fun build(component: ClusterComponent, @BindsInstance consumerGroup: ConsumerGroupId): ConsumerGroupComponent
+    }
+
+    fun cluster(): Cluster
+    fun consumerGroupId(): ConsumerGroupId
+    fun getConsumerGroupView(): ConsumerGroupView
+}

app/src/main/kotlin/insulator/di/components/InsulatorComponent.kt → app/src/main/kotlin/insulator/di/InsulatorComponent.kt

@@ -1,4 +1,4 @@
-package insulator.di.components
+package insulator.di
 
 import dagger.Component
 import insulator.configuration.ConfigurationRepo

app/src/main/kotlin/insulator/di/components/SubjectComponent.kt → app/src/main/kotlin/insulator/di/SubjectComponent.kt

@@ -1,12 +1,22 @@
-package insulator.di.components
+package insulator.di
 
 import dagger.BindsInstance
 import dagger.Component
-import insulator.di.SubjectScope
+import insulator.CachedFactory
 import insulator.kafka.model.Cluster
 import insulator.kafka.model.Subject
 import insulator.views.main.schemaregistry.CreateSchemaView
 import insulator.views.main.schemaregistry.SchemaView
+import javax.inject.Inject
+import javax.inject.Scope
+
+@Scope
+annotation class SubjectScope
+
+class SubjectComponentFactory @Inject constructor(clusterComponent: ClusterComponent) :
+    CachedFactory<Subject, SubjectComponent>({ subject: Subject ->
+        DaggerSubjectComponent.factory().build(clusterComponent, subject)
+    })
 
 @SubjectScope
 @Component(dependencies = [ClusterComponent::class])

app/src/main/kotlin/insulator/di/components/TopicComponent.kt → app/src/main/kotlin/insulator/di/TopicComponent.kt

@@ -1,15 +1,25 @@
-package insulator.di.components
+package insulator.di
 
 import dagger.BindsInstance
 import dagger.Component
-import insulator.di.TopicScope
+import insulator.CachedFactory
 import insulator.di.modules.TopicModule
 import insulator.jsonhelper.JsonFormatter
 import insulator.kafka.model.Cluster
 import insulator.kafka.model.Topic
 import insulator.views.main.topic.CreateTopicView
 import insulator.views.main.topic.ProducerView
 import insulator.views.main.topic.TopicView
+import javax.inject.Inject
+import javax.inject.Scope
+
+@Scope
+annotation class TopicScope
+
+class TopicComponentFactory @Inject constructor(clusterComponent: ClusterComponent) :
+    CachedFactory<Topic, TopicComponent>({ topic: Topic ->
+        DaggerTopicComponent.factory().build(clusterComponent, topic)
+    })
 
 @TopicScope
 @Component(dependencies = [ClusterComponent::class], modules = [TopicModule::class])

app/src/main/kotlin/insulator/ui/style/TreeViewStyle.kt

@@ -0,0 +1,40 @@
+package insulator.ui.style
+
+import tornadofx.Stylesheet
+import tornadofx.box
+import tornadofx.multi
+import tornadofx.px
+
+class TreeViewStyle : Stylesheet() {
+
+    init {
+        root {
+            treeView {
+                focusColor = theme.mainColor
+                backgroundColor = multi(theme.backgroundColor)
+                borderColor = multi(box(theme.backgroundColor))
+                backgroundInsets = multi(box(0.0.px))
+                padding = box(0.px, -theme.viewPadding)
+
+                treeCell {
+                    padding = box(10.0.px)
+                    borderRadius = multi(box(30.0.px))
+                    textFill = theme.black
+
+                    and(even) { backgroundColor = multi(theme.backgroundColor) }
+                    and(odd) { backgroundColor = multi(theme.backgroundColor) }
+                    and(hover) {
+                        backgroundColor = multi(theme.mainColor)
+                    }
+                    and(focused) {
+                        backgroundColor = multi(theme.mainColor)
+                    }
+                }
+
+                arrow {
+                    backgroundColor = multi(theme.black)
+                }
+            }
+        }
+    }
+}

app/src/main/kotlin/insulator/viewmodel/main/MainViewModel.kt

@@ -1,7 +1,7 @@
 package insulator.viewmodel.main
 
+import insulator.di.ClusterComponent
 import insulator.di.ClusterScope
-import insulator.di.components.ClusterComponent
 import insulator.kafka.model.Cluster
 import insulator.viewmodel.common.InsulatorViewModel
 import insulator.views.main.consumergroup.ListConsumerGroupView