You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jj...@apache.org on 2014/04/19 01:23:03 UTC
[1/2] git commit: KAFKA-1398 dynamic config changes are broken.
Repository: kafka
Updated Branches:
refs/heads/0.8.1 b18d2c379 -> 4bcb22f47
KAFKA-1398 dynamic config changes are broken.
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/2ce7ff6b
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/2ce7ff6b
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/2ce7ff6b
Branch: refs/heads/0.8.1
Commit: 2ce7ff6b6e56aa480792a298a8a92ef02e67a2f8
Parents: b18d2c3
Author: Jay Kreps <ja...@gmail.com>
Authored: Thu Apr 17 17:07:27 2014 -0700
Committer: Joel Koshy <jj...@gmail.com>
Committed: Fri Apr 18 16:15:55 2014 -0700
----------------------------------------------------------------------
.../scala/kafka/server/TopicConfigManager.scala | 52 ++++++++++++--------
.../kafka/server/DynamicConfigChangeTest.scala | 35 +++++++++++++
2 files changed, 67 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/2ce7ff6b/core/src/main/scala/kafka/server/TopicConfigManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/TopicConfigManager.scala b/core/src/main/scala/kafka/server/TopicConfigManager.scala
index d41fd33..4a4274e 100644
--- a/core/src/main/scala/kafka/server/TopicConfigManager.scala
+++ b/core/src/main/scala/kafka/server/TopicConfigManager.scala
@@ -59,7 +59,7 @@ import org.I0Itec.zkclient.{IZkChildListener, ZkClient}
*/
class TopicConfigManager(private val zkClient: ZkClient,
private val logManager: LogManager,
- private val changeExpirationMs: Long = 10*60*1000,
+ private val changeExpirationMs: Long = 15*60*1000,
private val time: Time = SystemTime) extends Logging {
private var lastExecutedChange = -1L
@@ -86,7 +86,7 @@ class TopicConfigManager(private val zkClient: ZkClient,
*/
private def processConfigChanges(notifications: Seq[String]) {
if (notifications.size > 0) {
- info("Processing %d topic config change notification(s)...".format(notifications.size))
+ info("Processing config change notification(s)...")
val now = time.milliseconds
val logs = logManager.logsByTopicPartition.toBuffer
val logsByTopic = logs.groupBy(_._1.topic).mapValues(_.map(_._2))
@@ -94,26 +94,38 @@ class TopicConfigManager(private val zkClient: ZkClient,
val changeId = changeNumber(notification)
if (changeId > lastExecutedChange) {
val changeZnode = ZkUtils.TopicConfigChangesPath + "/" + notification
- val (topicJson, stat) = ZkUtils.readData(zkClient, changeZnode)
- val topic = topicJson.substring(1, topicJson.length - 1) // dequote
- if (logsByTopic.contains(topic)) {
- /* combine the default properties with the overrides in zk to create the new LogConfig */
- val props = new Properties(logManager.defaultConfig.toProps)
- props.putAll(AdminUtils.fetchTopicConfig(zkClient, topic))
- val logConfig = LogConfig.fromProps(props)
- for (log <- logsByTopic(topic))
- log.config = logConfig
- lastExecutedChange = changeId
- info("Processed topic config change %d for topic %s, setting new config to %s.".format(changeId, topic, props))
- } else {
- if (now - stat.getCtime > changeExpirationMs) {
- /* this change is now obsolete, try to delete it unless it is the last change left */
- error("Ignoring topic config change %d for topic %s since the change has expired")
- } else {
- error("Ignoring topic config change %d for topic %s since the topic may have been deleted")
+ val (jsonOpt, stat) = ZkUtils.readDataMaybeNull(zkClient, ZkUtils.TopicConfigChangesPath + "/" + notification)
+ if(jsonOpt.isDefined) {
+ val json = jsonOpt.get
+ val topic = json.substring(1, json.length - 1) // hacky way to dequote
+ if (logsByTopic.contains(topic)) {
+ /* combine the default properties with the overrides in zk to create the new LogConfig */
+ val props = new Properties(logManager.defaultConfig.toProps)
+ props.putAll(AdminUtils.fetchTopicConfig(zkClient, topic))
+ val logConfig = LogConfig.fromProps(props)
+ for (log <- logsByTopic(topic))
+ log.config = logConfig
+ info("Processed topic config change %d for topic %s, setting new config to %s.".format(changeId, topic, props))
+ purgeObsoleteNotifications(now, notifications)
}
- ZkUtils.deletePath(zkClient, changeZnode)
}
+ lastExecutedChange = changeId
+ }
+ }
+ }
+ }
+
+ private def purgeObsoleteNotifications(now: Long, notifications: Seq[String]) {
+ val eligible = notifications.sorted.dropRight(1) // never purge newest notification--we need it for the seq number
+ for(notification <- eligible) {
+ val (jsonOpt, stat) = ZkUtils.readDataMaybeNull(zkClient, ZkUtils.TopicConfigChangesPath + "/" + notification)
+ if(jsonOpt.isDefined) {
+ val changeZnode = ZkUtils.TopicConfigChangesPath + "/" + notification
+ if (now - stat.getCtime > changeExpirationMs) {
+ debug("Purging config change notification " + notification)
+ ZkUtils.deletePath(zkClient, changeZnode)
+ } else {
+ return
}
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/2ce7ff6b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
new file mode 100644
index 0000000..5a1d5cc
--- /dev/null
+++ b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
@@ -0,0 +1,35 @@
+package kafka.server
+
+import junit.framework.Assert._
+import java.util.Properties
+import java.io.File
+import org.junit.{After, Before, Test}
+import kafka.integration.KafkaServerTestHarness
+import kafka.utils._
+import kafka.common._
+import kafka.log.LogConfig
+import kafka.admin.AdminUtils
+import org.scalatest.junit.JUnit3Suite
+
+class DynamicConfigChangeTest extends JUnit3Suite with KafkaServerTestHarness {
+
+ override val configs = List(new KafkaConfig(TestUtils.createBrokerConfig(0, TestUtils.choosePort)))
+
+ @Test
+ def testConfigChange() {
+ val oldVal = 100000
+ val newVal = 200000
+ val tp = TopicAndPartition("test", 0)
+ AdminUtils.createTopic(zkClient, tp.topic, 1, 1, LogConfig(flushInterval = oldVal).toProps)
+ TestUtils.retry(10000) {
+ val logOpt = this.servers(0).logManager.getLog(tp)
+ assertTrue(logOpt.isDefined)
+ assertEquals(oldVal, logOpt.get.config.flushInterval)
+ }
+ AdminUtils.changeTopicConfig(zkClient, tp.topic, LogConfig(flushInterval = newVal).toProps)
+ TestUtils.retry(10000) {
+ assertEquals(newVal, this.servers(0).logManager.getLog(tp).get.config.flushInterval)
+ }
+ }
+
+}
\ No newline at end of file
[2/2] git commit: KAFKA-1398 Dynamic config follow-on-comments.
Posted by jj...@apache.org.
KAFKA-1398 Dynamic config follow-on-comments.
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/4bcb22f4
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/4bcb22f4
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/4bcb22f4
Branch: refs/heads/0.8.1
Commit: 4bcb22f47e594c127b02d001620fd1f453034b2a
Parents: 2ce7ff6
Author: Jay Kreps <ja...@gmail.com>
Authored: Fri Apr 18 11:03:37 2014 -0700
Committer: Joel Koshy <jj...@gmail.com>
Committed: Fri Apr 18 16:16:02 2014 -0700
----------------------------------------------------------------------
.../scala/kafka/server/TopicConfigManager.scala | 6 +++---
.../unit/kafka/server/DynamicConfigChangeTest.scala | 16 ++++++++++++++++
2 files changed, 19 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/4bcb22f4/core/src/main/scala/kafka/server/TopicConfigManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/TopicConfigManager.scala b/core/src/main/scala/kafka/server/TopicConfigManager.scala
index 4a4274e..47295d4 100644
--- a/core/src/main/scala/kafka/server/TopicConfigManager.scala
+++ b/core/src/main/scala/kafka/server/TopicConfigManager.scala
@@ -40,6 +40,7 @@ import org.I0Itec.zkclient.{IZkChildListener, ZkClient}
* To update a topic config we first update the topic config properties. Then we create a new sequential
* znode under the change path which contains the name of the topic that was updated, say
* /brokers/config_changes/config_change_13321
+ * This is just a notification--the actual config change is stored only once under the /brokers/topics/<topic_name>/config path.
*
* This will fire a watcher on all brokers. This watcher works as follows. It reads all the config change notifications.
* It keeps track of the highest config change suffix number it has applied previously. For any previously applied change it finds
@@ -94,7 +95,7 @@ class TopicConfigManager(private val zkClient: ZkClient,
val changeId = changeNumber(notification)
if (changeId > lastExecutedChange) {
val changeZnode = ZkUtils.TopicConfigChangesPath + "/" + notification
- val (jsonOpt, stat) = ZkUtils.readDataMaybeNull(zkClient, ZkUtils.TopicConfigChangesPath + "/" + notification)
+ val (jsonOpt, stat) = ZkUtils.readDataMaybeNull(zkClient, changeZnode)
if(jsonOpt.isDefined) {
val json = jsonOpt.get
val topic = json.substring(1, json.length - 1) // hacky way to dequote
@@ -116,8 +117,7 @@ class TopicConfigManager(private val zkClient: ZkClient,
}
private def purgeObsoleteNotifications(now: Long, notifications: Seq[String]) {
- val eligible = notifications.sorted.dropRight(1) // never purge newest notification--we need it for the seq number
- for(notification <- eligible) {
+ for(notification <- notifications.sorted) {
val (jsonOpt, stat) = ZkUtils.readDataMaybeNull(zkClient, ZkUtils.TopicConfigChangesPath + "/" + notification)
if(jsonOpt.isDefined) {
val changeZnode = ZkUtils.TopicConfigChangesPath + "/" + notification
http://git-wip-us.apache.org/repos/asf/kafka/blob/4bcb22f4/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
index 5a1d5cc..5c48796 100644
--- a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
@@ -1,3 +1,19 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
package kafka.server
import junit.framework.Assert._