Merge pull request #23 from arenadata/feature/ADS-731

ADS-731. Added kafka 2.6.0 support.
arenadata · Jan 12, 2022 · 8357bc4 · 8357bc4
2 parents 0d16315 + 6a32fb7
commit 8357bc4
Show file tree

Hide file tree

Showing 8 changed files with 25 additions and 1 deletion.
diff --git a/app/controllers/Logkafka.scala b/app/controllers/Logkafka.scala
@@ -99,6 +99,8 @@ class Logkafka (val cc: ControllerComponents, val kafkaManagerContext: KafkaMana
     LogkafkaNewConfigs.configMaps(Kafka_2_2_0).map{case(k,v) => LKConfig(k,Some(v))}.toList)
   val kafka_2_4_0_Default = CreateLogkafka("","",
     LogkafkaNewConfigs.configMaps(Kafka_2_4_0).map{case(k,v) => LKConfig(k,Some(v))}.toList)
+  val kafka_2_6_0_Default = CreateLogkafka("","",
+    LogkafkaNewConfigs.configMaps(Kafka_2_6_0).map{case(k,v) => LKConfig(k,Some(v))}.toList)
   val kafka_2_8_1_Default = CreateLogkafka("","",
     LogkafkaNewConfigs.configMaps(Kafka_2_8_1).map{case(k,v) => LKConfig(k,Some(v))}.toList)
 
@@ -163,6 +165,7 @@ class Logkafka (val cc: ControllerComponents, val kafkaManagerContext: KafkaMana
           case Kafka_2_1_1 => (defaultCreateForm.fill(kafka_2_1_1_Default), clusterContext)
           case Kafka_2_2_0 => (defaultCreateForm.fill(kafka_2_2_0_Default), clusterContext)
           case Kafka_2_4_0 => (defaultCreateForm.fill(kafka_2_4_0_Default), clusterContext)
+          case Kafka_2_6_0 => (defaultCreateForm.fill(kafka_2_6_0_Default), clusterContext)
           case Kafka_2_8_1 => (defaultCreateForm.fill(kafka_2_8_1_Default), clusterContext)
         }
       }
@@ -269,6 +272,7 @@ class Logkafka (val cc: ControllerComponents, val kafkaManagerContext: KafkaMana
       case Kafka_2_1_1 => LogkafkaNewConfigs.configNames(Kafka_2_1_1).map(n => (n,LKConfig(n,None))).toMap
       case Kafka_2_2_0 => LogkafkaNewConfigs.configNames(Kafka_2_2_0).map(n => (n,LKConfig(n,None))).toMap
       case Kafka_2_4_0 => LogkafkaNewConfigs.configNames(Kafka_2_4_0).map(n => (n,LKConfig(n,None))).toMap
+      case Kafka_2_6_0 => LogkafkaNewConfigs.configNames(Kafka_2_6_0).map(n => (n,LKConfig(n,None))).toMap
       case Kafka_2_8_1 => LogkafkaNewConfigs.configNames(Kafka_2_8_1).map(n => (n,LKConfig(n,None))).toMap
     }
     val identityOption = li.identityMap.get(log_path)

diff --git a/app/controllers/Topic.scala b/app/controllers/Topic.scala
@@ -68,6 +68,7 @@ class Topic (val cc: ControllerComponents, val kafkaManagerContext: KafkaManager
   val kafka_2_1_1_Default = CreateTopic("",1,1,TopicConfigs.configNamesAndDoc(Kafka_2_1_1).map{ case (n, h) => TConfig(n,None,Option(h))}.toList)
   val kafka_2_2_0_Default = CreateTopic("",1,1,TopicConfigs.configNamesAndDoc(Kafka_2_2_0).map{ case (n, h) => TConfig(n,None,Option(h))}.toList)
   val kafka_2_4_0_Default = CreateTopic("",1,1,TopicConfigs.configNamesAndDoc(Kafka_2_4_0).map{ case (n, h) => TConfig(n,None,Option(h))}.toList)
+  val kafka_2_6_0_Default = CreateTopic("",1,1,TopicConfigs.configNamesAndDoc(Kafka_2_6_0).map{ case (n, h) => TConfig(n,None,Option(h))}.toList)
   val kafka_2_8_1_Default = CreateTopic("",1,1,TopicConfigs.configNamesAndDoc(Kafka_2_8_1).map{ case (n, h) => TConfig(n,None,Option(h))}.toList)
 
   val defaultCreateForm = Form(
@@ -172,6 +173,7 @@ class Topic (val cc: ControllerComponents, val kafkaManagerContext: KafkaManager
           case Kafka_2_1_1 => (defaultCreateForm.fill(kafka_2_1_1_Default), clusterContext)
           case Kafka_2_2_0 => (defaultCreateForm.fill(kafka_2_2_0_Default), clusterContext)
           case Kafka_2_4_0 => (defaultCreateForm.fill(kafka_2_4_0_Default), clusterContext)
+          case Kafka_2_6_0 => (defaultCreateForm.fill(kafka_2_6_0_Default), clusterContext)
           case Kafka_2_8_1 => (defaultCreateForm.fill(kafka_2_8_1_Default), clusterContext)
         }
       }
@@ -424,6 +426,7 @@ class Topic (val cc: ControllerComponents, val kafkaManagerContext: KafkaManager
           case Kafka_2_1_1 => TopicConfigs.configNamesAndDoc(Kafka_2_1_1).map { case (n, h) => (n,TConfig(n,None, Option(h))) }
           case Kafka_2_2_0 => TopicConfigs.configNamesAndDoc(Kafka_2_2_0).map { case (n, h) => (n,TConfig(n,None, Option(h))) }
           case Kafka_2_4_0 => TopicConfigs.configNamesAndDoc(Kafka_2_4_0).map { case (n, h) => (n,TConfig(n,None, Option(h))) }
+          case Kafka_2_6_0 => TopicConfigs.configNamesAndDoc(Kafka_2_6_0).map { case (n, h) => (n,TConfig(n,None, Option(h))) }
           case Kafka_2_8_1 => TopicConfigs.configNamesAndDoc(Kafka_2_8_1).map { case (n, h) => (n,TConfig(n,None, Option(h))) }
         }
         val updatedConfigMap = ti.config.toMap

diff --git a/app/kafka/manager/actor/cluster/KafkaStateActor.scala b/app/kafka/manager/actor/cluster/KafkaStateActor.scala
@@ -186,7 +186,7 @@ class KafkaAdminClient(context: => ActorContext, adminClientActorPath: ActorPath
 
 
 object KafkaManagedOffsetCache {
-  val supportedVersions: Set[KafkaVersion] = Set(Kafka_0_8_2_0, Kafka_0_8_2_1, Kafka_0_8_2_2, Kafka_0_9_0_0, Kafka_0_9_0_1, Kafka_0_10_0_0, Kafka_0_10_0_1, Kafka_0_10_1_0, Kafka_0_10_1_1, Kafka_0_10_2_0, Kafka_0_10_2_1, Kafka_0_11_0_0, Kafka_0_11_0_2, Kafka_1_0_0, Kafka_1_0_1, Kafka_1_1_0, Kafka_1_1_1, Kafka_2_0_0, Kafka_2_1_0, Kafka_2_1_1, Kafka_2_2_0, Kafka_2_4_0, Kafka_2_8_1)
+  val supportedVersions: Set[KafkaVersion] = Set(Kafka_0_8_2_0, Kafka_0_8_2_1, Kafka_0_8_2_2, Kafka_0_9_0_0, Kafka_0_9_0_1, Kafka_0_10_0_0, Kafka_0_10_0_1, Kafka_0_10_1_0, Kafka_0_10_1_1, Kafka_0_10_2_0, Kafka_0_10_2_1, Kafka_0_11_0_0, Kafka_0_11_0_2, Kafka_1_0_0, Kafka_1_0_1, Kafka_1_1_0, Kafka_1_1_1, Kafka_2_0_0, Kafka_2_1_0, Kafka_2_1_1, Kafka_2_2_0, Kafka_2_4_0, Kafka_2_6_0, Kafka_2_8_1)
   val ConsumerOffsetTopic = "__consumer_offsets"
 
   def isSupported(version: KafkaVersion) : Boolean = {

diff --git a/app/kafka/manager/model/model.scala b/app/kafka/manager/model/model.scala
@@ -104,6 +104,10 @@ case object Kafka_2_4_0 extends KafkaVersion {
   override def toString = "2.4.0"
 }
 
+case object Kafka_2_6_0 extends KafkaVersion {
+  override def toString = "2.6.0"
+}
+
 case object Kafka_2_8_1 extends KafkaVersion {
   override def toString = "2.8.1"
 }
@@ -134,6 +138,7 @@ object KafkaVersion {
     "2.1.1" -> Kafka_2_1_1,
     "2.2.0" -> Kafka_2_2_0,
     "2.4.0" -> Kafka_2_4_0,
+    "2.6.0" -> Kafka_2_6_0,
     "2.8.1" -> Kafka_2_8_1
   )
 

diff --git a/app/kafka/manager/utils/LogkafkaNewConfigs.scala b/app/kafka/manager/utils/LogkafkaNewConfigs.scala
@@ -41,6 +41,7 @@ object LogkafkaNewConfigs {
     Kafka_2_1_1 -> logkafka82.LogConfig,
     Kafka_2_2_0 -> logkafka82.LogConfig,
     Kafka_2_4_0 -> logkafka82.LogConfig,
+    Kafka_2_6_0 -> logkafka82.LogConfig,
     Kafka_2_8_1 -> logkafka82.LogConfig
   )
 

diff --git a/app/kafka/manager/utils/TopicConfigs.scala b/app/kafka/manager/utils/TopicConfigs.scala
@@ -44,6 +44,7 @@ object TopicConfigs {
     Kafka_2_1_1 -> two00.LogConfig,
     Kafka_2_2_0 -> two00.LogConfig,
     Kafka_2_4_0 -> two00.LogConfig,
+    Kafka_2_6_0 -> two00.LogConfig,
     Kafka_2_8_1 -> two00.LogConfig
   )
 

diff --git a/test/kafka/manager/model/KafkaVersionTest.scala b/test/kafka/manager/model/KafkaVersionTest.scala
@@ -36,6 +36,7 @@ class KafkaVersionTest extends FunSuite {
     "2.1.1" -> Kafka_2_1_1,
     "2.2.0" -> Kafka_2_2_0,
     "2.4.0" -> Kafka_2_4_0,
+    "2.6.0" -> Kafka_2_6_0,
     "2.8.1" -> Kafka_2_8_1
   )
 
@@ -79,6 +80,7 @@ class KafkaVersionTest extends FunSuite {
       ("2.1.1","2.1.1"),
       ("2.2.0","2.2.0"),
       ("2.4.0","2.4.0"),
+      ("2.6.0","2.6.0"),
       ("2.8.1","2.8.1")
     )
     assertResult(expected)(KafkaVersion.formSelectList)

diff --git a/test/kafka/manager/utils/TestClusterConfig.scala b/test/kafka/manager/utils/TestClusterConfig.scala
@@ -237,6 +237,14 @@ class TestClusterConfig extends FunSuite with Matchers {
     assert(cc == deserialize.get)
   }
 
+  test("serialize and deserialize 2.6.0") {
+    val cc = ClusterConfig("qa", "2.6.0", "localhost:2181", jmxEnabled = false, pollConsumers = true, filterConsumers = true, activeOffsetCacheEnabled = true, jmxUser = None, jmxPass = None, jmxSsl = false, tuning = None, securityProtocol = "SASL_PLAINTEXT", saslMechanism = Option("PLAIN"), jaasConfig = Option("blah"))
+    val serialize: String = ClusterConfig.serialize(cc)
+    val deserialize = ClusterConfig.deserialize(serialize)
+    assert(deserialize.isSuccess === true)
+    assert(cc == deserialize.get)
+  }
+
   test("serialize and deserialize 2.8.1") {
     val cc = ClusterConfig("qa", "2.8.1", "localhost:2181", jmxEnabled = false, pollConsumers = true, filterConsumers = true, activeOffsetCacheEnabled = true, jmxUser = None, jmxPass = None, jmxSsl = false, tuning = None, securityProtocol = "SASL_PLAINTEXT", saslMechanism = Option("PLAIN"), jaasConfig = Option("blah"))
     val serialize: String = ClusterConfig.serialize(cc)