- Remove group name parameter as not necessary

Author: Aliaksandr.Shpak

gradle.properties

@@ -1,3 +1,3 @@
-version = 0.1.1
+version = 0.1.2
 url = https://api.github.com/repos/alshpak/kafka_data_viewer/releases/latest
 update_url = https://github.com/alshpak/kafka_data_viewer/releases

src/main/scala/devtools/kafka_data_viewer/AppSettings.scala

@@ -47,9 +47,9 @@ object AppSettings {
     def withres[T <: AutoCloseable, R](t: T)(f: T => R): R = try f(t) finally t.close()
 
 
-    def connect(defaultGroup: String): AppSettings = {
+    def connect(): AppSettings = {
 
-        val $  = new DisposeStore()
+        val $ = new DisposeStore()
 
         val appPropsFile = new File("application.setting.yml")
         if (!appPropsFile.exists()) {
@@ -127,7 +127,6 @@ object AppSettings {
         root.listNodes[ConnectionDefinition]("connections", connections, ConnectionDefinition(), (item, node) => {
             node.property("name", item.name)
             node.property("kafkaHost", item.kafkaHost)
-            node.property("group", item.group, defaultGroup)
             node.listStrings("avroRegistries", item.avroRegistries)
             node.listCustom("topicsSettings", item.topicSettings)
         })

src/main/scala/devtools/kafka_data_viewer/KafkaDataViewer.scala

@@ -20,24 +20,21 @@ import devtools.lib.rxui.FxRender.DefaultFxRenderes
 import devtools.lib.rxui.UiImplicits._
 import devtools.lib.rxui._
 import io.reactivex.schedulers.Schedulers
-import org.apache.commons.cli.{DefaultParser, Options}
 import org.apache.kafka.common.errors.InterruptException
 
 import scala.concurrent.{ExecutionContextExecutor, Future}
 import scala.language.postfixOps
 
 class KafkaDataViewerAppPane(val layoutData: String = "",
                              connections: BehaviorSubject[Seq[ConnectionDefinition]],
-                             filters: BehaviorSubject[Seq[FilterData]],
-                             defaultGroup: String
+                             filters: BehaviorSubject[Seq[FilterData]]
                             )(implicit uiRenderer: UiRenderer) extends UiObservingComponent {
 
     case class ConnectionsSet(logging: ConsumerConnection, read: ConsumerConnection, master: ConsumerConnection, producer: ProducerConnection)
 
     private def connectToServices(connDef: ConnectionDefinition) = {
         val connector = new KafkaConnector(
-            host = connDef.kafkaHost.value,
-            groupName = connDef.group.value)
+            host = connDef.kafkaHost.value)
 
         ConnectionsSet(logging = connector.connectConsumer(), read = connector.connectConsumer(), master = connector.connectConsumer(), producer = connector.connectProducer())
     }
@@ -117,8 +114,7 @@ class KafkaDataViewerAppPane(val layoutData: String = "",
                     UiTab(label = "Connections List",
                         content = new KafkaConnectionsListPane(
                             connections = connections,
-                            onConnect = onConnect,
-                            defaultGroup = defaultGroup
+                            onConnect = onConnect
                         )))),
                 UiTabPanelExt[ConnHandle](
                     tabs = connectOps,
@@ -321,7 +317,6 @@ object KafkaDataViewer {
     case class ConnectionDefinition(
                                            name: BehaviorSubject[String] = behaviorSubject(""),
                                            kafkaHost: BehaviorSubject[String] = behaviorSubject(""),
-                                           group: BehaviorSubject[String] = behaviorSubject(""),
                                            topicSettings: BehaviorSubject[Seq[(String, MessageType)]] = behaviorSubject(Seq()),
                                            avroRegistries: BehaviorSubject[Seq[String]] = behaviorSubject(Seq()))
 
@@ -333,19 +328,18 @@ object KafkaDataViewer {
 
     def main(args: Array[String]): Unit = {
 
-        val options = new Options()
-        options.addOption("n", "groupname", true, "Group name to connect to kafka server")
-
-        val cli = new DefaultParser().parse(options, args)
-        val groupName = cli.getOptionValue("n")
-
-        val defaultGroup = Option(groupName).getOrElse("local.connection")
-        val appSettings = AppSettings.connect(defaultGroup = defaultGroup)
+        //        val options = new Options()
+        //        options.addOption("n", "groupname", true, "Group name to connect to kafka server")
+        //
+        //        val cli = new DefaultParser().parse(options, args)
+        //        val groupName = cli.getOptionValue("n")
+        //
+        //        val defaultGroup = Option(groupName).getOrElse("local.connection")
+        val appSettings = AppSettings.connect()
 
         DefaultFxRenderes.runApp(root = new KafkaDataViewerAppPane(
             connections = appSettings.connections,
-            filters = appSettings.filters,
-            defaultGroup = defaultGroup
+            filters = appSettings.filters
         ), postAction = uiRenderer => {
             implicit val ec: ExecutionContextExecutor = scala.concurrent.ExecutionContext.global
             Future { AppUpdate.verify() }.onComplete(r => {

src/main/scala/devtools/kafka_data_viewer/kafkaconn/KafkaConnector.scala

@@ -1,7 +1,7 @@
 package devtools.kafka_data_viewer.kafkaconn
 
 import java.time.Duration.ofSeconds
-import java.time.{Duration, Instant}
+import java.time.Instant
 import java.util
 import java.util.concurrent.Semaphore
 import java.util.concurrent.atomic.AtomicInteger
@@ -13,29 +13,28 @@ import devtools.lib.rxui.DisposeStore
 import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord, KafkaConsumer}
 import org.apache.kafka.clients.producer.{KafkaProducer, Partitioner, ProducerConfig, ProducerRecord}
 import org.apache.kafka.common.errors.InterruptException
-import org.apache.kafka.common.serialization.{ByteArrayDeserializer, ByteArraySerializer, StringSerializer}
+import org.apache.kafka.common.serialization.{ByteArrayDeserializer, ByteArraySerializer}
 import org.apache.kafka.common.{Cluster, PartitionInfo, TopicPartition}
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 import scala.language.postfixOps
 
-class KafkaConnector(host: String,
-                     groupName: String) extends Connector {
+class KafkaConnector(host: String) extends Connector {
 
 
     private val consumers = ArrayBuffer[ConsumerConnection]()
     private val producers = ArrayBuffer[ProducerConnection]()
 
     override def connectConsumer(): ConsumerConnection = {
-        val connection = new KafkaConsumerConnection(host, groupName)
+        val connection = new KafkaConsumerConnection(host)
         consumers += connection
         connection
     }
 
     override def connectProducer(): ProducerConnection = {
-        val connection = new KafkaProducerConnection(host, groupName)
+        val connection = new KafkaProducerConnection(host)
         producers += connection
         connection
     }
@@ -51,12 +50,10 @@ object Counter {
     val counter = new AtomicInteger(0)
 }
 
-class KafkaConsumerConnection(host: String,
-                              groupName: String) extends ConsumerConnection {
+class KafkaConsumerConnection(host: String) extends ConsumerConnection {
 
     private val consumerProps = Map(
         "bootstrap.servers" -> host
-        , "group.id" -> groupName
         , "key.deserializer" -> classOf[ByteArrayDeserializer].getName
         , "value.deserializer" -> classOf[ByteArrayDeserializer].getName
         , "enable.auto.commit" -> "false"
@@ -140,8 +137,8 @@ class KafkaConsumerConnection(host: String,
             }
 
             val nextRecords: Unit => Seq[GenRecord] = _ =>
-                if (assignment.isEmpty) {try Thread.sleep(1000) catch { case e: InterruptedException =>  }; Seq() }
-                else try consumer.poll(ofSeconds(1)).asScala.toSeq catch { case e: InterruptException => Seq() }
+                if (assignment.isEmpty) {try Thread.sleep(1000) catch {case e: InterruptedException =>}; Seq() }
+                else try consumer.poll(ofSeconds(1)).asScala.toSeq catch {case e: InterruptException => Seq()}
 
 
             for (records <- Stream.continually(reassign.andThen(nextRecords)(getNewTopics())).takeWhile(_ => !closed))
@@ -203,8 +200,7 @@ class KafkaConsumerConnection(host: String,
 
 }
 
-class KafkaProducerConnection(host: String,
-                              groupName: String) extends ProducerConnection {
+class KafkaProducerConnection(host: String) extends ProducerConnection {
 
     private val producerProps = Map(
         ProducerConfig.BOOTSTRAP_SERVERS_CONFIG -> host,
@@ -217,9 +213,9 @@ class KafkaProducerConnection(host: String,
 
     override def send(topic: String, key: Array[Byte], value: Array[Byte], partition: PartitionerMode): Unit = {
         try {
-        println("Try to send message!")
-        CustomPartitioner.partitionMode = partition
-        producer.send(new ProducerRecord(topic, key, value))
+            println("Try to send message!")
+            CustomPartitioner.partitionMode = partition
+            producer.send(new ProducerRecord(topic, key, value))
         } catch {
             case e: Exception => e.printStackTrace()
         }

src/main/scala/devtools/kafka_data_viewer/ui/KafkaConnectionsListPane.scala

@@ -11,8 +11,7 @@ import scala.language.postfixOps
 
 class KafkaConnectionsListPane(val layoutData: String = "",
                                connections: BehaviorSubject[Seq[ConnectionDefinition]],
-                               onConnect: Subject[ConnectionDefinition],
-                               defaultGroup: String
+                               onConnect: Subject[ConnectionDefinition]
                               )(implicit uiRenderer: UiRenderer) extends UiObservingComponent {
 
     private val onAdd = publishSubject[Unit]()
@@ -24,7 +23,6 @@ class KafkaConnectionsListPane(val layoutData: String = "",
         val applyHandle = publishSubject[Unit]()
         val closeHandle = publishSubject[Unit]()
         val newConnection = ConnectionDefinition()
-        newConnection.group << defaultGroup
         for (conn <- $(applyHandle)) connections << connections.value :+ newConnection
         uiRenderer.runModal(new ConfigureConnectionWindow("", newConnection, applyHandle, closeHandle), close = closeHandle)
     }
@@ -72,24 +70,20 @@ class ConfigureConnectionWindow(
     for (_ <- $(onOk)) {
         conn.name << name.value.trim
         conn.kafkaHost << host.value.trim
-        conn.group << group.value.trim
         onApply onNext Unit
         onClose onNext Unit
     }
     private val name = behaviorSubject(conn.name.value)
     private val host = behaviorSubject[String](conn.kafkaHost.value)
-    private val group = behaviorSubject[String](conn.group.value)
 
-    private val applyAllowed = Observable.combineLatest(name, host, group)
-            .map(x => !x._1.trim.isEmpty && !x._2.trim.isEmpty && !x._3.trim.isEmpty)
+    private val applyAllowed = Observable.combineLatest(name, host)
+            .map(x => !x._1.trim.isEmpty && !x._2.trim.isEmpty)
 
     override def content(): UiWidget = UiPanel(layoutData, Grid("cols 2,margin 5"), items = Seq(
         UiLabel(text = "Connection Name"),
         UiText("growx", text = name),
         UiLabel(text = "Kafka Host"),
         UiText("growx", text = host),
-        UiLabel(text = "Consumer Group"),
-        UiText("growx", text = group),
         UiSeparator("colspan 2, growx, graby, valign T", orientation = UiHoriz),
         UiPanel("colspan 2, halign R", Grid("cols 2,margin 2"), items = Seq(
             UiButton(text = "Cancel", onAction = onClose, cancelButton = true),