diff --git a/app/views/broker/updateConfig.scala.html b/app/views/broker/updateConfig.scala.html
new file mode 100644
index 000000000..9eadeac85
--- /dev/null
+++ b/app/views/broker/updateConfig.scala.html
@@ -0,0 +1,57 @@
+@*
+* Copyright 2015 Yahoo Inc. Licensed under the Apache License, Version 2.0
+* See accompanying LICENSE file.
+*@
+@import scalaz.{\/}
+@(cluster: String, brokerId: Int, errorOrForm: kafka.manager.ApiError \/ (Form[models.form.UpdateBrokerConfig], String)
+)(implicit af: features.ApplicationFeatures, messages: play.api.i18n.Messages, menus: models.navigation.Menus, request:RequestHeader)
+
+@import helper._
+@import controllers.routes
+
+
+@theMenu = {
+ @views.html.navigation.clusterMenu(cluster,"Broker","Update Config",menus.clusterMenus(cluster)(
+ kafka.manager.features.ClusterFeatures.default))
+}
+
+@renderForm(updateBrokerConfigForm: Form[models.form.UpdateBrokerConfig]) = {
+ @b4.vertical.form(routes.Cluster.handleUpdateBrokerConfig(cluster, brokerId)) { implicit fc =>
+
+
+ Update Config
+
notice: replication.throttled.rate need set topic level replication.throttled.replicas to work. |
|---|
+
+
+
+ |
+ @b4.text(updateBrokerConfigForm("broker"), '_label -> "Broker", 'placeholder -> "", 'autofocus -> true )
+ @b4.hidden(updateBrokerConfigForm("readVersion").name,updateBrokerConfigForm("readVersion").value.getOrElse(-1))
+ @helper.repeat(updateBrokerConfigForm("configs"), min = 1) { configsForm =>
+ @b4.hidden(configsForm("name").name, configsForm("name").value.getOrElse(""))
+ @b4.hidden(configsForm("help").name, configsForm("help").value.getOrElse(""))
+ @b4.text(configsForm("value"), '_label -> configsForm("name").value.getOrElse(""), '_help -> configsForm("help").value.getOrElse(""))
+ }
+ |
+
+
+
+ @b4.submit('class -> "submit-button btn btn-primary"){ Update Config }
+
Cancel
+ }
+}
+
+@main(
+ "Update Config",
+ menu = theMenu,
+ breadcrumbs=views.html.navigation.breadCrumbs(models.navigation.BreadCrumbs.withNamedViewAndClusterAndBroker("Broker View",cluster,brokerId,"Update Config"))) {
+
+
+
+
+ @errorOrForm.fold( views.html.errors.onApiError(_), t => renderForm(t._1))
+
+
+
+}
+
diff --git a/conf/routes b/conf/routes
index 27ec9de98..ba8b1e70c 100644
--- a/conf/routes
+++ b/conf/routes
@@ -16,6 +16,8 @@ GET /clusters/:c/logkafkas controllers.Logkafka
GET /clusters/:c/logkafkas/:h/:l controllers.Logkafka.logkafka(c:String, h:String, l:String)
GET /clusters/:c/brokers controllers.Cluster.brokers(c: String)
GET /clusters/:c/brokers/:b controllers.Cluster.broker(c: String, b:Int)
+GET /clusters/:c/brokers/:b/updateConfig controllers.Cluster.updateBrokerConfig(c:String,b: Int)
+POST /clusters/:c/brokers/:b/updateConfig controllers.Cluster.handleUpdateBrokerConfig(c:String,b: Int)
GET /clusters/:c/consumers controllers.Consumer.consumers(c: String)
GET /clusters/:c/consumers/:g/type/:ct controllers.Consumer.consumer(c: String, g:String, ct: String)
GET /clusters/:c/consumers/:g/topic/:t/type/:ct controllers.Consumer.consumerAndTopic(c: String, g:String, t:String, ct: String)
diff --git a/test/kafka/manager/TestKafkaManager.scala b/test/kafka/manager/TestKafkaManager.scala
index ca070c319..94fe80e41 100644
--- a/test/kafka/manager/TestKafkaManager.scala
+++ b/test/kafka/manager/TestKafkaManager.scala
@@ -391,6 +391,32 @@ class TestKafkaManager extends CuratorAwareTest with BaseTest {
}
}
+ test("update broker config") {
+ val tiFuture= kafkaManager.getBrokerIdentity("dev",0)
+ val tiOrError = Await.result(tiFuture, duration)
+ assert(tiOrError.isRight, "Failed to get broker identity!")
+ val ti = tiOrError.toOption.get
+ val config = new Properties()
+ config.put(kafka.manager.utils.zero11.BrokerConfig.FollowerReplicationThrottledRateProp,"10000000")
+ val configReadVersion = ti.configReadVersion
+ val future = kafkaManager.updateBrokerConfig("dev",0,config,configReadVersion)
+ val result = Await.result(future,duration)
+ assert(result.isRight === true)
+ Thread.sleep(2000)
+
+ //check new config
+ {
+ val tiFuture= kafkaManager.getBrokerIdentity("dev",0)
+ val tiOrError = Await.result(tiFuture, duration)
+ assert(tiOrError.isRight, "Failed to get broker identity!")
+ val ti = tiOrError.toOption.get
+ assert(ti.configReadVersion > configReadVersion)
+ assert(ti.config.toMap.apply(kafka.manager.utils.zero11.BrokerConfig.FollowerReplicationThrottledRateProp) === "10000000")
+ }
+ }
+
+
+
test("delete topic") {
val futureA = kafkaManager.deleteTopic("dev",createTopicNameA)
val resultA = Await.result(futureA,duration)