You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2012/07/27 01:56:59 UTC
svn commit: r1366238 -
/incubator/kafka/branches/0.8/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala
Author: junrao
Date: Thu Jul 26 23:56:59 2012
New Revision: 1366238
URL: http://svn.apache.org/viewvc?rev=1366238&view=rev
Log:
NullPointerException in ConsumerFetcherManager; patched by Jun Rao; reviewed by Neha Narkhede; KAFKA-418
Modified:
incubator/kafka/branches/0.8/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala
Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala?rev=1366238&r1=1366237&r2=1366238&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala Thu Jul 26 23:56:59 2012
@@ -107,14 +107,18 @@ class ConsumerFetcherManager(private val
}
def stopAllConnections() {
+ // first, clear noLeaderPartitionSet so that no more fetchers can be added to leader_finder_thread
lock.lock()
- try {
- partitionMap = null
- noLeaderPartitionSet.clear()
- } finally {
- lock.unlock()
- }
+ noLeaderPartitionSet.clear()
+ lock.unlock()
+
+ // second, stop all existing fetchers
closeAllFetchers()
+
+ // finally clear partitionMap
+ lock.lock()
+ partitionMap = null
+ lock.unlock()
}
def getPartitionTopicInfo(key: (String, Int)) : PartitionTopicInfo = {