kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject [7/37] git commit: KAFKA-749 Bug in socket server shutdown logic makes the broker hang on shutdown until it has to be killed; reviewed by Sriram and Jay Kreps
Date Mon, 04 Mar 2013 04:22:03 GMT
KAFKA-749 Bug in socket server shutdown logic makes the broker hang on shutdown until it has
to be killed; reviewed by Sriram and Jay Kreps


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

Branch: refs/heads/trunk
Commit: 826f02a74e00bec1efb9d60b43cf9e0702d8d58a
Parents: aed6c3c
Author: Neha Narkhede <neha.narkhede@gmail.com>
Authored: Tue Feb 5 20:11:03 2013 -0800
Committer: Neha Narkhede <neha.narkhede@gmail.com>
Committed: Tue Feb 5 20:11:03 2013 -0800

----------------------------------------------------------------------
 .../main/scala/kafka/network/RequestChannel.scala  |    4 ++++
 .../main/scala/kafka/network/SocketServer.scala    |    1 +
 .../scala/kafka/server/KafkaRequestHandler.scala   |    2 +-
 core/src/main/scala/kafka/server/KafkaServer.scala |    4 ++--
 4 files changed, 8 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/826f02a7/core/src/main/scala/kafka/network/RequestChannel.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala
index 9b0f7e9..7747ddd 100644
--- a/core/src/main/scala/kafka/network/RequestChannel.scala
+++ b/core/src/main/scala/kafka/network/RequestChannel.scala
@@ -126,6 +126,10 @@ class RequestChannel(val numProcessors: Int, val queueSize: Int) extends
KafkaMe
   def addResponseListener(onResponse: Int => Unit) { 
     responseListeners ::= onResponse
   }
+
+  def shutdown() {
+    requestQueue.clear
+  }
 }
 
 object RequestMetrics {

http://git-wip-us.apache.org/repos/asf/kafka/blob/826f02a7/core/src/main/scala/kafka/network/SocketServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala
index b056e25..8f0053a 100644
--- a/core/src/main/scala/kafka/network/SocketServer.scala
+++ b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -71,6 +71,7 @@ class SocketServer(val brokerId: Int,
       acceptor.shutdown()
     for(processor <- processors)
       processor.shutdown()
+    requestChannel.shutdown
     info("shut down completely")
   }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/826f02a7/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
index f0c05a5..69ca058 100644
--- a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
+++ b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
@@ -32,7 +32,7 @@ class KafkaRequestHandler(id: Int, brokerId: Int, val requestChannel: RequestCha
     while(true) {
       try {
         val req = requestChannel.receiveRequest()
-        if(req eq RequestChannel.AllDone){
+        if(req eq RequestChannel.AllDone) {
           trace("receives shut down command, shut down".format(brokerId, id))
           return
         }

http://git-wip-us.apache.org/repos/asf/kafka/blob/826f02a7/core/src/main/scala/kafka/server/KafkaServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index 1fe1ca9..09e261f 100644
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -110,6 +110,8 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime) extends
Logg
     info("shutting down")
     val canShutdown = isShuttingDown.compareAndSet(false, true);
     if (canShutdown) {
+      if(socketServer != null)
+        Utils.swallow(socketServer.shutdown())
       if(requestHandlerPool != null)
         Utils.swallow(requestHandlerPool.shutdown())
       Utils.swallow(kafkaScheduler.shutdown())
@@ -119,8 +121,6 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime) extends
Logg
         Utils.swallow(kafkaZookeeper.shutdown())
       if(replicaManager != null)
         Utils.swallow(replicaManager.shutdown())
-      if(socketServer != null)
-        Utils.swallow(socketServer.shutdown())
       if(logManager != null)
         Utils.swallow(logManager.shutdown())
 


Mime
View raw message