kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jkr...@apache.org
Subject [5/30] git commit: KAFKA-646 Missing files from previous commit
Date Tue, 18 Dec 2012 17:44:12 GMT
KAFKA-646 Missing files from previous commit


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/994fe081
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/994fe081
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/994fe081

Branch: refs/heads/trunk
Commit: 994fe081b6e429ba1cfd5c09d464015c41e64cf7
Parents: 792b071
Author: Swapnil Ghike <sghike@linkedin.com>
Authored: Mon Dec 17 15:53:12 2012 -0800
Committer: Neha Narkhede <nehanarkhede@apache.org>
Committed: Mon Dec 17 15:53:12 2012 -0800

----------------------------------------------------------------------
 .../scala/kafka/common/ClientIdAndBroker.scala     |   26 ++++
 .../main/scala/kafka/common/ClientIdAndTopic.scala |   27 +++++
 core/src/main/scala/kafka/common/Config.scala      |   40 +++++++
 core/src/main/scala/kafka/common/Topic.scala       |   40 +++++++
 .../consumer/FetchRequestAndResponseStats.scala    |   58 ++++++++++
 .../kafka/producer/ProducerRequestStats.scala      |   56 +++++++++
 .../main/scala/kafka/producer/ProducerStats.scala  |   40 +++++++
 .../scala/kafka/producer/ProducerTopicStats.scala  |   57 +++++++++
 .../test/scala/unit/kafka/common/ConfigTest.scala  |   89 +++++++++++++++
 .../test/scala/unit/kafka/common/TopicTest.scala   |   61 ++++++++++
 10 files changed, 494 insertions(+), 0 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/main/scala/kafka/common/ClientIdAndBroker.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/common/ClientIdAndBroker.scala b/core/src/main/scala/kafka/common/ClientIdAndBroker.scala
new file mode 100644
index 0000000..93223a9
--- /dev/null
+++ b/core/src/main/scala/kafka/common/ClientIdAndBroker.scala
@@ -0,0 +1,26 @@
+package kafka.common
+
+/**
+ * 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.
+ */
+
+/**
+ * Convenience case class since (clientId, brokerInfo) pairs are used to create
+ * SyncProducer Request Stats and SimpleConsumer Request and Response Stats.
+ */
+case class ClientIdAndBroker(clientId: String, brokerInfo: String) {
+  override def toString = "%s-%s".format(clientId, brokerInfo)
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/main/scala/kafka/common/ClientIdAndTopic.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/common/ClientIdAndTopic.scala b/core/src/main/scala/kafka/common/ClientIdAndTopic.scala
new file mode 100644
index 0000000..7acf9e7
--- /dev/null
+++ b/core/src/main/scala/kafka/common/ClientIdAndTopic.scala
@@ -0,0 +1,27 @@
+package kafka.common
+
+/**
+ * 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.
+ */
+
+/**
+ * Convenience case class since (clientId, topic) pairs are used in the creation
+ * of many Stats objects.
+ */
+case class ClientIdAndTopic(clientId: String, topic: String) {
+  override def toString = "%s-%s".format(clientId, topic)
+}
+

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/main/scala/kafka/common/Config.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/common/Config.scala b/core/src/main/scala/kafka/common/Config.scala
new file mode 100644
index 0000000..53bfcfb
--- /dev/null
+++ b/core/src/main/scala/kafka/common/Config.scala
@@ -0,0 +1,40 @@
+/**
+ * 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.common
+
+import util.matching.Regex
+import kafka.utils.Logging
+
+trait Config extends Logging {
+
+  def validateChars(prop: String, value: String) {
+    val legalChars = "[a-zA-Z0-9_-]"
+    val rgx = new Regex(legalChars + "*")
+
+    rgx.findFirstIn(value) match {
+      case Some(t) =>
+        if (!t.equals(value))
+          throw new InvalidConfigException(prop + " " + value + " is illegal, contains a
character other than ASCII alphanumerics, _ and -")
+      case None => throw new InvalidConfigException(prop + " " + value + " is illegal,
contains a character other than ASCII alphanumerics, _ and -")
+    }
+  }
+}
+
+
+
+

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/main/scala/kafka/common/InvalidClientIdException.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/common/InvalidClientIdException.scala b/core/src/main/scala/kafka/common/InvalidClientIdException.scala
deleted file mode 100644
index e69de29..0000000

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/main/scala/kafka/common/Topic.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/common/Topic.scala b/core/src/main/scala/kafka/common/Topic.scala
new file mode 100644
index 0000000..c96ed62
--- /dev/null
+++ b/core/src/main/scala/kafka/common/Topic.scala
@@ -0,0 +1,40 @@
+/**
+ * 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.common
+
+import util.matching.Regex
+
+object Topic {
+  private val legalChars = "[a-zA-Z0-9_-]"
+  private val maxNameLength = 255
+  private val rgx = new Regex(legalChars + "+")
+
+  def validate(topic: String) {
+    if (topic.length <= 0)
+      throw new InvalidTopicException("topic name is illegal, can't be empty")
+    else if (topic.length > maxNameLength)
+      throw new InvalidTopicException("topic name is illegal, can't be longer than " + maxNameLength
+ " characters")
+
+    rgx.findFirstIn(topic) match {
+      case Some(t) =>
+        if (!t.equals(topic))
+          throw new InvalidTopicException("topic name " + topic + " is illegal, contains
a character other than ASCII alphanumerics, _ and -")
+      case None => throw new InvalidTopicException("topic name " + topic + " is illegal,
 contains a character other than ASCII alphanumerics, _ and -")
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/main/scala/kafka/consumer/FetchRequestAndResponseStats.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/consumer/FetchRequestAndResponseStats.scala b/core/src/main/scala/kafka/consumer/FetchRequestAndResponseStats.scala
new file mode 100644
index 0000000..2cc0f36
--- /dev/null
+++ b/core/src/main/scala/kafka/consumer/FetchRequestAndResponseStats.scala
@@ -0,0 +1,58 @@
+/**
+ * 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.consumer
+
+import kafka.metrics.{KafkaTimer, KafkaMetricsGroup}
+import kafka.utils.Pool
+import java.util.concurrent.TimeUnit
+import kafka.common.ClientIdAndBroker
+
+class FetchRequestAndResponseMetrics(metricId: ClientIdAndBroker) extends KafkaMetricsGroup
{
+  val requestTimer = new KafkaTimer(newTimer(metricId + "-FetchRequestRateAndTimeMs", TimeUnit.MILLISECONDS,
TimeUnit.SECONDS))
+  val requestSizeHist = newHistogram(metricId + "-FetchResponseSize")
+}
+
+/**
+ * Tracks metrics of the requests made by a given consumer client to all brokers, and the
responses obtained from the brokers.
+ * @param clientId ClientId of the given consumer
+ */
+class FetchRequestAndResponseStats(clientId: String) {
+  private val valueFactory = (k: ClientIdAndBroker) => new FetchRequestAndResponseMetrics(k)
+  private val stats = new Pool[ClientIdAndBroker, FetchRequestAndResponseMetrics](Some(valueFactory))
+  private val allBrokersStats = new FetchRequestAndResponseMetrics(new ClientIdAndBroker(clientId,
"All.Brokers"))
+
+  def getFetchRequestAndResponseAllBrokersStats(): FetchRequestAndResponseMetrics = allBrokersStats
+
+  def getFetchRequestAndResponseStats(brokerInfo: String): FetchRequestAndResponseMetrics
= {
+    stats.getAndMaybePut(new ClientIdAndBroker(clientId, brokerInfo))
+  }
+}
+
+/**
+ * Stores the fetch request and response stats information of each consumer client in a (clientId
-> FetchRequestAndResponseStats) map.
+ */
+object FetchRequestAndResponseStatsRegistry {
+  private val valueFactory = (k: String) => new FetchRequestAndResponseStats(k)
+  private val globalStats = new Pool[String, FetchRequestAndResponseStats](Some(valueFactory))
+
+  def getFetchRequestAndResponseStats(clientId: String) = {
+    globalStats.getAndMaybePut(clientId)
+  }
+}
+
+

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/main/scala/kafka/producer/ProducerRequestStats.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/producer/ProducerRequestStats.scala b/core/src/main/scala/kafka/producer/ProducerRequestStats.scala
new file mode 100644
index 0000000..e29ccad
--- /dev/null
+++ b/core/src/main/scala/kafka/producer/ProducerRequestStats.scala
@@ -0,0 +1,56 @@
+/**
+ * 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.producer
+
+import kafka.metrics.{KafkaTimer, KafkaMetricsGroup}
+import java.util.concurrent.TimeUnit
+import kafka.utils.Pool
+import kafka.common.ClientIdAndBroker
+
+class ProducerRequestMetrics(metricId: ClientIdAndBroker) extends KafkaMetricsGroup {
+  val requestTimer = new KafkaTimer(newTimer(metricId + "-ProducerRequestRateAndTimeMs",
TimeUnit.MILLISECONDS, TimeUnit.SECONDS))
+  val requestSizeHist = newHistogram(metricId + "-ProducerRequestSize")
+}
+
+/**
+ * Tracks metrics of requests made by a given producer client to all brokers.
+ * @param clientId ClientId of the given producer
+ */
+class ProducerRequestStats(clientId: String) {
+  private val valueFactory = (k: ClientIdAndBroker) => new ProducerRequestMetrics(k)
+  private val stats = new Pool[ClientIdAndBroker, ProducerRequestMetrics](Some(valueFactory))
+  private val allBrokersStats = new ProducerRequestMetrics(new ClientIdAndBroker(clientId,
"All.Brokers"))
+
+  def getProducerRequestAllBrokersStats(): ProducerRequestMetrics = allBrokersStats
+
+  def getProducerRequestStats(brokerInfo: String): ProducerRequestMetrics = {
+    stats.getAndMaybePut(new ClientIdAndBroker(clientId, brokerInfo))
+  }
+}
+
+/**
+ * Stores the request stats information of each producer client in a (clientId -> ProducerRequestStats)
map.
+ */
+object ProducerRequestStatsRegistry {
+  private val valueFactory = (k: String) => new ProducerRequestStats(k)
+  private val globalStats = new Pool[String, ProducerRequestStats](Some(valueFactory))
+
+  def getProducerRequestStats(clientId: String) = {
+    globalStats.getAndMaybePut(clientId)
+  }
+}
+

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/main/scala/kafka/producer/ProducerStats.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/producer/ProducerStats.scala b/core/src/main/scala/kafka/producer/ProducerStats.scala
new file mode 100644
index 0000000..62ff544
--- /dev/null
+++ b/core/src/main/scala/kafka/producer/ProducerStats.scala
@@ -0,0 +1,40 @@
+/**
+ * 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.producer
+
+import kafka.metrics.KafkaMetricsGroup
+import java.util.concurrent.TimeUnit
+import kafka.utils.Pool
+
+class ProducerStats(clientId: String) extends KafkaMetricsGroup {
+  val serializationErrorRate = newMeter(clientId + "-SerializationErrorsPerSec",  "errors",
TimeUnit.SECONDS)
+  val resendRate = newMeter(clientId + "-ResendsPerSec",  "resends", TimeUnit.SECONDS)
+  val failedSendRate = newMeter(clientId + "-FailedSendsPerSec",  "failed sends", TimeUnit.SECONDS)
+  val droppedMessageRate = newMeter(clientId + "-DroppedMessagesPerSec",  "drops", TimeUnit.SECONDS)
+}
+
+/**
+ * Stores metrics of serialization and message sending activity of each producer client in
a (clientId -> ProducerStats) map.
+ */
+object ProducerStatsRegistry {
+  private val valueFactory = (k: String) => new ProducerStats(k)
+  private val statsRegistry = new Pool[String, ProducerStats](Some(valueFactory))
+
+  def getProducerStats(clientId: String) = {
+    statsRegistry.getAndMaybePut(clientId)
+  }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/main/scala/kafka/producer/ProducerTopicStats.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/producer/ProducerTopicStats.scala b/core/src/main/scala/kafka/producer/ProducerTopicStats.scala
new file mode 100644
index 0000000..fd0b44e
--- /dev/null
+++ b/core/src/main/scala/kafka/producer/ProducerTopicStats.scala
@@ -0,0 +1,57 @@
+/**
+ * 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.producer
+
+import kafka.metrics.KafkaMetricsGroup
+import kafka.common.ClientIdAndTopic
+import kafka.utils.{Pool, threadsafe}
+import java.util.concurrent.TimeUnit
+
+
+@threadsafe
+class ProducerTopicMetrics(metricId: ClientIdAndTopic) extends KafkaMetricsGroup {
+  val messageRate = newMeter(metricId + "-MessagesPerSec",  "messages", TimeUnit.SECONDS)
+  val byteRate = newMeter(metricId + "-BytesPerSec",  "bytes", TimeUnit.SECONDS)
+}
+
+/**
+ * Tracks metrics for each topic the given producer client has produced data to.
+ * @param clientId The clientId of the given producer client.
+ */
+class ProducerTopicStats(clientId: String) {
+  private val valueFactory = (k: ClientIdAndTopic) => new ProducerTopicMetrics(k)
+  private val stats = new Pool[ClientIdAndTopic, ProducerTopicMetrics](Some(valueFactory))
+  private val allTopicStats = new ProducerTopicMetrics(new ClientIdAndTopic(clientId, "All.Topics"))
// to differentiate from a topic named AllTopics
+
+  def getProducerAllTopicStats(): ProducerTopicMetrics = allTopicStats
+
+  def getProducerTopicStats(topic: String): ProducerTopicMetrics = {
+    stats.getAndMaybePut(new ClientIdAndTopic(clientId, topic))
+  }
+}
+
+/**
+ * Stores the topic stats information of each producer client in a (clientId -> ProducerTopicStats)
map.
+ */
+object ProducerTopicStatsRegistry {
+  private val valueFactory = (k: String) => new ProducerTopicStats(k)
+  private val globalStats = new Pool[String, ProducerTopicStats](Some(valueFactory))
+
+  def getProducerTopicStats(clientId: String) = {
+    globalStats.getAndMaybePut(clientId)
+  }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/main/scala/kafka/utils/ClientIdAndTopic.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/utils/ClientIdAndTopic.scala b/core/src/main/scala/kafka/utils/ClientIdAndTopic.scala
deleted file mode 100644
index e69de29..0000000

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/test/scala/unit/kafka/common/ConfigTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/common/ConfigTest.scala b/core/src/test/scala/unit/kafka/common/ConfigTest.scala
new file mode 100644
index 0000000..6226dda
--- /dev/null
+++ b/core/src/test/scala/unit/kafka/common/ConfigTest.scala
@@ -0,0 +1,89 @@
+/**
+ * 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 unit.kafka.common
+
+import junit.framework.Assert._
+import collection.mutable.ArrayBuffer
+import org.junit.Test
+import kafka.common.InvalidConfigException
+import kafka.producer.ProducerConfig
+import kafka.consumer.ConsumerConfig
+
+class ConfigTest {
+
+  @Test
+  def testInvalidClientIds() {
+    val invalidClientIds = new ArrayBuffer[String]()
+    val badChars = Array('/', '\\', ',', '\0', ':', "\"", '\'', ';', '*', '?', '.', ' ',
'\t', '\r', '\n', '=')
+    for (weirdChar <- badChars) {
+      invalidClientIds += "Is" + weirdChar + "illegal"
+    }
+
+    for (i <- 0 until invalidClientIds.size) {
+      try {
+        ProducerConfig.validateClientId(invalidClientIds(i))
+        fail("Should throw InvalidClientIdException.")
+      }
+      catch {
+        case e: InvalidConfigException => "This is good."
+      }
+    }
+
+    val validClientIds = new ArrayBuffer[String]()
+    validClientIds += ("valid", "CLIENT", "iDs", "ar6", "VaL1d", "_0-9_", "")
+    for (i <- 0 until validClientIds.size) {
+      try {
+        ProducerConfig.validateClientId(validClientIds(i))
+      }
+      catch {
+        case e: Exception => fail("Should not throw exception.")
+      }
+    }
+  }
+
+  @Test
+  def testInvalidGroupIds() {
+    val invalidGroupIds = new ArrayBuffer[String]()
+    val badChars = Array('/', '\\', ',', '\0', ':', "\"", '\'', ';', '*', '?', '.', ' ',
'\t', '\r', '\n', '=')
+    for (weirdChar <- badChars) {
+      invalidGroupIds += "Is" + weirdChar + "illegal"
+    }
+
+    for (i <- 0 until invalidGroupIds.size) {
+      try {
+        ConsumerConfig.validateGroupId(invalidGroupIds(i))
+        fail("Should throw InvalidGroupIdException.")
+      }
+      catch {
+        case e: InvalidConfigException => "This is good."
+      }
+    }
+
+    val validGroupIds = new ArrayBuffer[String]()
+    validGroupIds += ("valid", "GROUP", "iDs", "ar6", "VaL1d", "_0-9_", "")
+    for (i <- 0 until validGroupIds.size) {
+      try {
+        ConsumerConfig.validateGroupId(validGroupIds(i))
+      }
+      catch {
+        case e: Exception => fail("Should not throw exception.")
+      }
+    }
+  }
+}
+

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/test/scala/unit/kafka/common/TopicTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/common/TopicTest.scala b/core/src/test/scala/unit/kafka/common/TopicTest.scala
new file mode 100644
index 0000000..b37553e
--- /dev/null
+++ b/core/src/test/scala/unit/kafka/common/TopicTest.scala
@@ -0,0 +1,61 @@
+/**
+ * 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 unit.kafka.common
+
+import junit.framework.Assert._
+import collection.mutable.ArrayBuffer
+import kafka.common.{Topic, InvalidTopicException}
+import org.junit.Test
+
+class TopicTest {
+
+  @Test
+  def testInvalidTopicNames() {
+    val invalidTopicNames = new ArrayBuffer[String]()
+    invalidTopicNames += ("", ".", "..")
+    var longName = "ATCG"
+    for (i <- 1 to 6)
+      longName += longName
+    invalidTopicNames += longName
+    val badChars = Array('/', '\\', ',', '\0', ':', "\"", '\'', ';', '*', '?', '.')
+    for (weirdChar <- badChars) {
+      invalidTopicNames += "Is" + weirdChar + "illegal"
+    }
+
+    for (i <- 0 until invalidTopicNames.size) {
+      try {
+        Topic.validate(invalidTopicNames(i))
+        fail("Should throw InvalidTopicException.")
+      }
+      catch {
+        case e: InvalidTopicException => "This is good."
+      }
+    }
+
+    val validTopicNames = new ArrayBuffer[String]()
+    validTopicNames += ("valid", "TOPIC", "nAmEs", "ar6", "VaL1d", "_0-9_")
+    for (i <- 0 until validTopicNames.size) {
+      try {
+        Topic.validate(validTopicNames(i))
+      }
+      catch {
+        case e: Exception => fail("Should not throw exception.")
+      }
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/test/scala/unit/kafka/utils/ClientIdTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/utils/ClientIdTest.scala b/core/src/test/scala/unit/kafka/utils/ClientIdTest.scala
deleted file mode 100644
index e69de29..0000000

http://git-wip-us.apache.org/repos/asf/kafka/blob/994fe081/core/src/test/scala/unit/kafka/utils/TopicTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/utils/TopicTest.scala b/core/src/test/scala/unit/kafka/utils/TopicTest.scala
deleted file mode 100644
index e69de29..0000000


Mime
View raw message