Skip to content

Commit

Permalink
Merge pull request #28 from arenadata/feature/kafka-3.6.2
Browse files Browse the repository at this point in the history
KAFKA 3.6.2
  • Loading branch information
Asmoday authored Aug 27, 2024
2 parents 0ed9f9d + 13b44af commit 514ed5f
Show file tree
Hide file tree
Showing 8 changed files with 30 additions and 6 deletions.
4 changes: 4 additions & 0 deletions app/controllers/Logkafka.scala
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,8 @@ class Logkafka (val cc: ControllerComponents, val kafkaManagerContext: KafkaMana
LogkafkaNewConfigs.configMaps(Kafka_2_8_1).map{case(k,v) => LKConfig(k,Some(v))}.toList)
val kafka_3_3_2_Default = CreateLogkafka("", "",
LogkafkaNewConfigs.configMaps(Kafka_3_3_2).map { case (k, v) => LKConfig(k, Some(v)) }.toList)
val kafka_3_6_2_Default = CreateLogkafka("", "",
LogkafkaNewConfigs.configMaps(Kafka_3_6_2).map { case (k, v) => LKConfig(k, Some(v)) }.toList)

val defaultCreateForm = Form(
mapping(
Expand Down Expand Up @@ -169,6 +171,7 @@ class Logkafka (val cc: ControllerComponents, val kafkaManagerContext: KafkaMana
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)
case Kafka_3_3_2 => (defaultCreateForm.fill(kafka_3_3_2_Default), clusterContext)
case Kafka_3_6_2 => (defaultCreateForm.fill(kafka_3_6_2_Default), clusterContext)
}
}
}
Expand Down Expand Up @@ -277,6 +280,7 @@ class Logkafka (val cc: ControllerComponents, val kafkaManagerContext: KafkaMana
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
case Kafka_3_3_2 => LogkafkaNewConfigs.configNames(Kafka_3_3_2).map(n => (n,LKConfig(n,None))).toMap
case Kafka_3_6_2 => LogkafkaNewConfigs.configNames(Kafka_3_6_2).map(n => (n,LKConfig(n,None))).toMap
}
val identityOption = li.identityMap.get(log_path)
if (identityOption.isDefined) {
Expand Down
3 changes: 3 additions & 0 deletions app/controllers/Topic.scala
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ class Topic (val cc: ControllerComponents, val kafkaManagerContext: KafkaManager
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 kafka_3_3_2_Default = CreateTopic("",1,1,TopicConfigs.configNamesAndDoc(Kafka_3_3_2).map{ case (n, h) => TConfig(n,None,Option(h))}.toList)
val kafka_3_6_2_Default = CreateTopic("",1,1,TopicConfigs.configNamesAndDoc(Kafka_3_6_2).map{ case (n, h) => TConfig(n,None,Option(h))}.toList)


val defaultCreateForm = Form(
Expand Down Expand Up @@ -178,6 +179,7 @@ class Topic (val cc: ControllerComponents, val kafkaManagerContext: KafkaManager
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)
case Kafka_3_3_2 => (defaultCreateForm.fill(kafka_3_3_2_Default), clusterContext)
case Kafka_3_6_2 => (defaultCreateForm.fill(kafka_3_6_2_Default), clusterContext)
}
}
}
Expand Down Expand Up @@ -432,6 +434,7 @@ class Topic (val cc: ControllerComponents, val kafkaManagerContext: KafkaManager
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))) }
case Kafka_3_3_2 => TopicConfigs.configNamesAndDoc(Kafka_3_3_2).map { case (n, h) => (n,TConfig(n,None, Option(h))) }
case Kafka_3_6_2 => TopicConfigs.configNamesAndDoc(Kafka_3_6_2).map { case (n, h) => (n,TConfig(n,None, Option(h))) }
}
val updatedConfigMap = ti.config.toMap
val updatedConfigList = defaultConfigs.map {
Expand Down
2 changes: 1 addition & 1 deletion app/kafka/manager/actor/cluster/KafkaStateActor.scala
Original file line number Diff line number Diff line change
Expand Up @@ -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_6_0, Kafka_2_8_1, Kafka_3_3_2)
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, Kafka_3_3_2, Kafka_3_6_2)
val ConsumerOffsetTopic = "__consumer_offsets"

def isSupported(version: KafkaVersion) : Boolean = {
Expand Down
7 changes: 6 additions & 1 deletion app/kafka/manager/model/model.scala
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,10 @@ case object Kafka_3_3_2 extends KafkaVersion {
override def toString: String = "3.3.2"
}

case object Kafka_3_6_2 extends KafkaVersion {
override def toString: String = "3.6.2"
}

object KafkaVersion {
val supportedVersions: Map[String,KafkaVersion] = Map(
"0.8.1.1" -> Kafka_0_8_1_1,
Expand Down Expand Up @@ -144,7 +148,8 @@ object KafkaVersion {
"2.4.0" -> Kafka_2_4_0,
"2.6.0" -> Kafka_2_6_0,
"2.8.1" -> Kafka_2_8_1,
"3.3.2" -> Kafka_3_3_2
"3.3.2" -> Kafka_3_3_2,
"3.6.2" -> Kafka_3_6_2
)

val formSelectList : IndexedSeq[(String,String)] = supportedVersions.toIndexedSeq.filterNot(_._1.contains("beta")).map(t => (t._1,t._2.toString)).sortWith((a, b) => sortVersion(a._1, b._1))
Expand Down
3 changes: 2 additions & 1 deletion app/kafka/manager/utils/LogkafkaNewConfigs.scala
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,8 @@ object LogkafkaNewConfigs {
Kafka_2_4_0 -> logkafka82.LogConfig,
Kafka_2_6_0 -> logkafka82.LogConfig,
Kafka_2_8_1 -> logkafka82.LogConfig,
Kafka_3_3_2 -> logkafka82.LogConfig
Kafka_3_3_2 -> logkafka82.LogConfig,
Kafka_3_6_2 -> logkafka82.LogConfig
)

def configNames(version: KafkaVersion) : Set[String] = {
Expand Down
3 changes: 2 additions & 1 deletion app/kafka/manager/utils/TopicConfigs.scala
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,8 @@ object TopicConfigs {
Kafka_2_4_0 -> two00.LogConfig,
Kafka_2_6_0 -> two00.LogConfig,
Kafka_2_8_1 -> two00.LogConfig,
Kafka_3_3_2 -> two00.LogConfig
Kafka_3_3_2 -> two00.LogConfig,
Kafka_3_6_2 -> two00.LogConfig
)

def configNames(version: KafkaVersion): Seq[String] = {
Expand Down
6 changes: 4 additions & 2 deletions test/kafka/manager/model/KafkaVersionTest.scala
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,8 @@ class KafkaVersionTest extends FunSuite {
"2.4.0" -> Kafka_2_4_0,
"2.6.0" -> Kafka_2_6_0,
"2.8.1" -> Kafka_2_8_1,
"3.3.2" -> Kafka_3_3_2
"3.3.2" -> Kafka_3_3_2,
"3.6.2" -> Kafka_3_6_2
)

test("apply method: supported version.") {
Expand Down Expand Up @@ -83,7 +84,8 @@ class KafkaVersionTest extends FunSuite {
("2.4.0","2.4.0"),
("2.6.0","2.6.0"),
("2.8.1","2.8.1"),
("3.3.2", "3.3.2")
("3.3.2", "3.3.2"),
("3.6.2", "3.6.2")
)
assertResult(expected)(KafkaVersion.formSelectList)
}
Expand Down
8 changes: 8 additions & 0 deletions test/kafka/manager/utils/TestClusterConfig.scala
Original file line number Diff line number Diff line change
Expand Up @@ -260,4 +260,12 @@ class TestClusterConfig extends FunSuite with Matchers {
assert(deserialize.isSuccess === true)
assert(cc == deserialize.get)
}

test("serialize and deserialize 3.6.2") {
val cc = ClusterConfig("qa", "3.6.2", "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)
}
}

0 comments on commit 514ed5f

Please sign in to comment.