kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nehanarkh...@apache.org
Subject git commit: KAFKA-690 TopicMetadataRequest throws exception when no topics are specified, reviewed by Jay Kreps and Neha Narkhede
Date Wed, 09 Jan 2013 16:49:14 GMT
Updated Branches:
  refs/heads/0.8 85c9e91c8 -> e0b3b6316


KAFKA-690 TopicMetadataRequest throws exception when no topics are specified, reviewed by
Jay Kreps and Neha Narkhede


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

Branch: refs/heads/0.8
Commit: e0b3b6316342271d798af10494c0ceab13e1eb02
Parents: 85c9e91
Author: Neha Narkhede <neha.narkhede@gmail.com>
Authored: Wed Jan 9 08:49:11 2013 -0800
Committer: Neha Narkhede <neha.narkhede@gmail.com>
Committed: Wed Jan 9 08:49:11 2013 -0800

----------------------------------------------------------------------
 .../scala/kafka/api/TopicMetadataRequest.scala     |    1 -
 core/src/main/scala/kafka/server/KafkaApis.scala   |    9 +++-
 .../unit/kafka/integration/TopicMetadataTest.scala |   31 ++++++++++++---
 3 files changed, 32 insertions(+), 9 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/e0b3b631/core/src/main/scala/kafka/api/TopicMetadataRequest.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/api/TopicMetadataRequest.scala b/core/src/main/scala/kafka/api/TopicMetadataRequest.scala
index 5bdb2c1..e659532 100644
--- a/core/src/main/scala/kafka/api/TopicMetadataRequest.scala
+++ b/core/src/main/scala/kafka/api/TopicMetadataRequest.scala
@@ -40,7 +40,6 @@ object TopicMetadataRequest extends Logging {
     for(i <- 0 until numTopics)
       topics += readShortString(buffer)
     val topicsList = topics.toList
-    debug("topic = %s".format(topicsList.head))
     new TopicMetadataRequest(versionId, clientId, topics.toList, correlationId)
   }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/e0b3b631/core/src/main/scala/kafka/server/KafkaApis.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index 5a85b04..5089a75 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -21,7 +21,6 @@ import kafka.admin.{CreateTopicCommand, AdminUtils}
 import kafka.api._
 import kafka.message._
 import kafka.network._
-import kafka.utils.{Pool, SystemTime, Logging}
 import org.apache.log4j.Logger
 import scala.collection._
 import kafka.network.RequestChannel.Response
@@ -30,6 +29,7 @@ import java.util.concurrent.atomic._
 import kafka.metrics.KafkaMetricsGroup
 import org.I0Itec.zkclient.ZkClient
 import kafka.common._
+import kafka.utils.{ZkUtils, Pool, SystemTime, Logging}
 
 
 /**
@@ -425,7 +425,12 @@ class KafkaApis(val requestChannel: RequestChannel,
 
     val topicsMetadata = new mutable.ArrayBuffer[TopicMetadata]()
     val config = replicaManager.config
-    val uniqueTopics = metadataRequest.topics.toSet
+    val uniqueTopics = {
+      if(metadataRequest.topics.size > 0)
+        metadataRequest.topics.toSet
+      else
+        ZkUtils.getAllTopics(zkClient).toSet
+    }
     val topicMetadataList = AdminUtils.fetchTopicMetadataFromZk(uniqueTopics, zkClient)
     topicMetadataList.foreach(
       topicAndMetadata => {

http://git-wip-us.apache.org/repos/asf/kafka/blob/e0b3b631/core/src/test/scala/unit/kafka/integration/TopicMetadataTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/integration/TopicMetadataTest.scala b/core/src/test/scala/unit/kafka/integration/TopicMetadataTest.scala
index 54a5a06..f9bbfa9 100644
--- a/core/src/test/scala/unit/kafka/integration/TopicMetadataTest.scala
+++ b/core/src/test/scala/unit/kafka/integration/TopicMetadataTest.scala
@@ -70,7 +70,27 @@ class TopicMetadataTest extends JUnit3Suite with ZooKeeperTestHarness {
       0 -> configs.head.brokerId
     )
     TestUtils.makeLeaderForPartition(zkClient, topic, leaderForPartitionMap, 1)
-    val topicMetadata = mockLogManagerAndTestTopic(topic)
+    val topicMetadataRequest = new TopicMetadataRequest(List(topic))
+    val topicMetadata = mockLogManagerAndTestTopic(topicMetadataRequest)
+    assertEquals("Expecting metadata only for 1 topic", 1, topicMetadata.size)
+    assertEquals("Expecting metadata for the test topic", "test", topicMetadata.head.topic)
+    val partitionMetadata = topicMetadata.head.partitionsMetadata
+    assertEquals("Expecting metadata for 1 partition", 1, partitionMetadata.size)
+    assertEquals("Expecting partition id to be 0", 0, partitionMetadata.head.partitionId)
+    assertEquals(1, partitionMetadata.head.replicas.size)
+  }
+
+  def testGetAllTopicMetadata {
+    // create topic
+    val topic = "test"
+    CreateTopicCommand.createTopic(zkClient, topic, 1)
+    // set up leader for topic partition 0
+    val leaderForPartitionMap = Map(
+      0 -> configs.head.brokerId
+    )
+    TestUtils.makeLeaderForPartition(zkClient, topic, leaderForPartitionMap, 1)
+    val topicMetadataRequest = new TopicMetadataRequest(List())
+    val topicMetadata = mockLogManagerAndTestTopic(topicMetadataRequest)
     assertEquals("Expecting metadata only for 1 topic", 1, topicMetadata.size)
     assertEquals("Expecting metadata for the test topic", "test", topicMetadata.head.topic)
     val partitionMetadata = topicMetadata.head.partitionsMetadata
@@ -83,7 +103,8 @@ class TopicMetadataTest extends JUnit3Suite with ZooKeeperTestHarness {
     // auto create topic
     val topic = "test"
 
-    val topicMetadata = mockLogManagerAndTestTopic(topic)
+    val topicMetadataRequest = new TopicMetadataRequest(List(topic))
+    val topicMetadata = mockLogManagerAndTestTopic(topicMetadataRequest)
     assertEquals("Expecting metadata only for 1 topic", 1, topicMetadata.size)
     assertEquals("Expecting metadata for the test topic", "test", topicMetadata.head.topic)
     val partitionMetadata = topicMetadata.head.partitionsMetadata
@@ -94,16 +115,14 @@ class TopicMetadataTest extends JUnit3Suite with ZooKeeperTestHarness
{
     assertEquals(ErrorMapping.LeaderNotAvailableCode, partitionMetadata.head.errorCode)
   }
 
-  private def mockLogManagerAndTestTopic(topic: String): Seq[TopicMetadata] = {
+  private def mockLogManagerAndTestTopic(request: TopicMetadataRequest): Seq[TopicMetadata]
= {
     // topic metadata request only requires 1 call from the replica manager
     val replicaManager = EasyMock.createMock(classOf[ReplicaManager])
     EasyMock.expect(replicaManager.config).andReturn(configs.head).anyTimes()
     EasyMock.replay(replicaManager)
 
-    // create a topic metadata request
-    val topicMetadataRequest = new TopicMetadataRequest(List(topic))
 
-    val serializedMetadataRequest = TestUtils.createRequestByteBuffer(topicMetadataRequest)
+    val serializedMetadataRequest = TestUtils.createRequestByteBuffer(request)
 
     // create the kafka request handler
     val requestChannel = new RequestChannel(2, 5)


Mime
View raw message