kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nehanarkh...@apache.org
Subject git commit: KAFKA-1317 follow up fix
Date Sat, 29 Mar 2014 01:21:53 GMT
Repository: kafka
Updated Branches:
  refs/heads/0.8.1 39a560789 -> 5a6a1d83b


KAFKA-1317 follow up fix


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

Branch: refs/heads/0.8.1
Commit: 5a6a1d83b81caaebec0f10a0039b422acaeb5c35
Parents: 39a5607
Author: Neha Narkhede <neha.narkhede@gmail.com>
Authored: Fri Mar 28 18:21:47 2014 -0700
Committer: Neha Narkhede <neha.narkhede@gmail.com>
Committed: Fri Mar 28 18:21:47 2014 -0700

----------------------------------------------------------------------
 .../scala/kafka/controller/TopicDeletionManager.scala    | 11 +++++------
 1 file changed, 5 insertions(+), 6 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/5a6a1d83/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/TopicDeletionManager.scala b/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
index 6f615cf..40c4c57 100644
--- a/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
+++ b/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
@@ -23,6 +23,7 @@ import collection.Set
 import kafka.common.{ErrorMapping, TopicAndPartition}
 import kafka.api.{StopReplicaResponse, RequestOrResponse}
 import java.util.concurrent.locks.ReentrantLock
+import java.util.concurrent.atomic.AtomicBoolean
 
 /**
  * This manages the state machine for topic deletion.
@@ -76,7 +77,7 @@ class TopicDeletionManager(controller: KafkaController,
   var topicsIneligibleForDeletion: mutable.Set[String] = mutable.Set.empty[String] ++
     (initialTopicsIneligibleForDeletion & initialTopicsToBeDeleted)
   val deleteTopicsCond = deleteLock.newCondition()
-  var deleteTopicStateChanged: Boolean = false
+  var deleteTopicStateChanged: AtomicBoolean = new AtomicBoolean(false)
   var deleteTopicsThread: DeleteTopicsThread = null
   val isDeleteTopicEnabled = controller.config.deleteTopicEnable
 
@@ -86,7 +87,7 @@ class TopicDeletionManager(controller: KafkaController,
   def start() {
     if(isDeleteTopicEnabled) {
       deleteTopicsThread = new DeleteTopicsThread()
-      deleteTopicStateChanged = true
+      deleteTopicStateChanged.set(true)
       deleteTopicsThread.start()
     }
   }
@@ -198,12 +199,10 @@ class TopicDeletionManager(controller: KafkaController,
    */
   private def awaitTopicDeletionNotification() {
     inLock(deleteLock) {
-      while(!deleteTopicStateChanged) {
+      while(!deleteTopicStateChanged.compareAndSet(true, false)) {
         info("Waiting for signal to start or continue topic deletion")
-
         deleteTopicsCond.await()
       }
-      deleteTopicStateChanged = false
     }
   }
 
@@ -211,7 +210,7 @@ class TopicDeletionManager(controller: KafkaController,
    * Signals the delete-topic-thread to process topic deletion
    */
   private def resumeTopicDeletionThread() {
-    deleteTopicStateChanged = true
+    deleteTopicStateChanged.set(true)
     inLock(deleteLock) {
       deleteTopicsCond.signal()
     }


Mime
View raw message