kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nehanarkh...@apache.org
Subject git commit: KAFKA-1030 Addition of partitions requires bouncing all the consumers of that topic; reviewed by Neha, Swapnil, Joel
Date Tue, 17 Sep 2013 21:22:53 GMT
Updated Branches:
  refs/heads/0.8 aebf74619 -> c6ca97173


KAFKA-1030 Addition of partitions requires bouncing all the consumers of that topic; reviewed
by Neha, Swapnil, Joel


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

Branch: refs/heads/0.8
Commit: c6ca971738700643ecba78ad1f3998062481aab9
Parents: aebf746
Author: Neha Narkhede <neha.narkhede@gmail.com>
Authored: Tue Sep 17 14:22:48 2013 -0700
Committer: Neha Narkhede <neha.narkhede@gmail.com>
Committed: Tue Sep 17 14:22:48 2013 -0700

----------------------------------------------------------------------
 .../kafka/consumer/ZookeeperConsumerConnector.scala   | 14 ++------------
 1 file changed, 2 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/c6ca9717/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala b/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
index 81bf0bd..881f51e 100644
--- a/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
+++ b/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
@@ -31,7 +31,6 @@ import java.util.UUID
 import kafka.serializer._
 import kafka.utils.ZkUtils._
 import kafka.common._
-import kafka.client.ClientUtils
 import com.yammer.metrics.core.Gauge
 import kafka.metrics._
 import scala.Some
@@ -422,17 +421,8 @@ private[kafka] class ZookeeperConsumerConnector(val config: ConsumerConfig,
         true
       }
       else {
-        val topicsMetadata = ClientUtils.fetchTopicMetadata(myTopicThreadIdsMap.keySet,
-                                                            brokers,
-                                                            config.clientId,
-                                                            config.socketTimeoutMs,
-                                                            correlationId.getAndIncrement).topicsMetadata
-        val partitionsPerTopicMap = new mutable.HashMap[String, Seq[Int]]
-        topicsMetadata.foreach(m => {
-          val topic = m.topic
-          val partitions = m.partitionsMetadata.map(m1 => m1.partitionId)
-          partitionsPerTopicMap.put(topic, partitions)
-        })
+        val partitionsAssignmentPerTopicMap = getPartitionAssignmentForTopics(zkClient, myTopicThreadIdsMap.keySet.toSeq)
+        val partitionsPerTopicMap = partitionsAssignmentPerTopicMap.map(p => (p._1, p._2.keySet.toSeq))
 
         /**
          * fetchers must be stopped to avoid data duplication, since if the current


Mime
View raw message