incubator-kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject svn commit: r1213990 - in /incubator/kafka/trunk/core/src/main/scala/kafka: common/ConsumerReblanceFailedException.scala consumer/ZookeeperConsumerConnector.scala consumer/ZookeeperTopicEventWatcher.scala server/KafkaServerStartable.scala
Date Tue, 13 Dec 2011 23:55:53 GMT
Author: junrao
Date: Tue Dec 13 23:55:52 2011
New Revision: 1213990

URL: http://svn.apache.org/viewvc?rev=1213990&view=rev
Log:
Bug in mirroring code causes mirroring to halt; patched by Jun Rao; reviewed by Neha Narkhede;
KAFKA-225

Added:
    incubator/kafka/trunk/core/src/main/scala/kafka/common/ConsumerReblanceFailedException.scala
Modified:
    incubator/kafka/trunk/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
    incubator/kafka/trunk/core/src/main/scala/kafka/consumer/ZookeeperTopicEventWatcher.scala
    incubator/kafka/trunk/core/src/main/scala/kafka/server/KafkaServerStartable.scala

Added: incubator/kafka/trunk/core/src/main/scala/kafka/common/ConsumerReblanceFailedException.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/trunk/core/src/main/scala/kafka/common/ConsumerReblanceFailedException.scala?rev=1213990&view=auto
==============================================================================
--- incubator/kafka/trunk/core/src/main/scala/kafka/common/ConsumerReblanceFailedException.scala
(added)
+++ incubator/kafka/trunk/core/src/main/scala/kafka/common/ConsumerReblanceFailedException.scala
Tue Dec 13 23:55:52 2011
@@ -0,0 +1,26 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.common
+
+/**
+ * Thrown when a request is made for broker but no brokers with that topic
+ * exist.
+ */
+class ConsumerRebalanceFailedException(message: String) extends RuntimeException(message)
{
+  def this() = this(null)
+}
\ No newline at end of file

Modified: incubator/kafka/trunk/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/trunk/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala?rev=1213990&r1=1213989&r2=1213990&view=diff
==============================================================================
--- incubator/kafka/trunk/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
(original)
+++ incubator/kafka/trunk/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
Tue Dec 13 23:55:52 2011
@@ -29,7 +29,7 @@ import org.apache.zookeeper.Watcher.Even
 import kafka.api.OffsetRequest
 import java.util.UUID
 import kafka.serializer.Decoder
-import kafka.common.InvalidConfigException
+import kafka.common.{ConsumerRebalanceFailedException, InvalidConfigException}
 
 /**
  * This class handles the consumers interaction with zookeeper
@@ -446,7 +446,7 @@ private[kafka] class ZookeeperConsumerCo
         }
       }
 
-      throw new RuntimeException(consumerIdString + " can't rebalance after " + config.maxRebalanceRetries
+" retries")
+      throw new ConsumerRebalanceFailedException(consumerIdString + " can't rebalance after
" + config.maxRebalanceRetries +" retries")
     }
 
     private def rebalance(): Boolean = {

Modified: incubator/kafka/trunk/core/src/main/scala/kafka/consumer/ZookeeperTopicEventWatcher.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/trunk/core/src/main/scala/kafka/consumer/ZookeeperTopicEventWatcher.scala?rev=1213990&r1=1213989&r2=1213990&view=diff
==============================================================================
--- incubator/kafka/trunk/core/src/main/scala/kafka/consumer/ZookeeperTopicEventWatcher.scala
(original)
+++ incubator/kafka/trunk/core/src/main/scala/kafka/consumer/ZookeeperTopicEventWatcher.scala
Tue Dec 13 23:55:52 2011
@@ -21,9 +21,11 @@ import scala.collection.JavaConversions.
 import kafka.utils.{Utils, ZkUtils, ZKStringSerializer, Logging}
 import org.I0Itec.zkclient.{IZkStateListener, IZkChildListener, ZkClient}
 import org.apache.zookeeper.Watcher.Event.KeeperState
+import kafka.server.KafkaServerStartable
+import kafka.common.ConsumerRebalanceFailedException
 
 class ZookeeperTopicEventWatcher(val config:ConsumerConfig,
-    val eventHandler: TopicEventHandler[String]) extends Logging {
+    val eventHandler: TopicEventHandler[String], kafkaServerStartable: KafkaServerStartable)
extends Logging {
 
   val lock = new Object()
 
@@ -33,7 +35,7 @@ class ZookeeperTopicEventWatcher(val con
   startWatchingTopicEvents()
 
   private def startWatchingTopicEvents() {
-    val topicEventListener = new ZkTopicEventListener
+    val topicEventListener = new ZkTopicEventListener(kafkaServerStartable)
     ZkUtils.makeSurePersistentPathExists(zkClient, ZkUtils.BrokerTopicsPath)
 
     zkClient.subscribeStateChanges(
@@ -50,24 +52,17 @@ class ZookeeperTopicEventWatcher(val con
 
   def shutdown() {
     lock.synchronized {
-      try {
-        if (zkClient != null) {
-          stopWatchingTopicEvents()
-          zkClient.close()
-          zkClient = null
-        }
-        else
-          warn("Cannot shutdown already shutdown topic event watcher.")
-      }
-      catch {
-        case e =>
-          fatal(e)
-          fatal(Utils.stackTrace(e))
+      if (zkClient != null) {
+        stopWatchingTopicEvents()
+        zkClient.close()
+        zkClient = null
       }
+      else
+        warn("Cannot shutdown already shutdown topic event watcher.")
     }
   }
 
-  class ZkTopicEventListener() extends IZkChildListener {
+  class ZkTopicEventListener(val kafkaServerStartable: KafkaServerStartable) extends IZkChildListener
{
 
     @throws(classOf[Exception])
     def handleChildChange(parent: String, children: java.util.List[String]) {
@@ -81,9 +76,11 @@ class ZookeeperTopicEventWatcher(val con
           }
         }
         catch {
+          case e: ConsumerRebalanceFailedException =>
+            fatal("can't rebalance in embedded consumer; proceed to shutdown", e)
+            kafkaServerStartable.shutdown()
           case e =>
-            fatal(e)
-            fatal(Utils.stackTrace(e))
+            error("error in handling child changes in embedded consumer", e)
         }
       }
     }

Modified: incubator/kafka/trunk/core/src/main/scala/kafka/server/KafkaServerStartable.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/trunk/core/src/main/scala/kafka/server/KafkaServerStartable.scala?rev=1213990&r1=1213989&r2=1213990&view=diff
==============================================================================
--- incubator/kafka/trunk/core/src/main/scala/kafka/server/KafkaServerStartable.scala (original)
+++ incubator/kafka/trunk/core/src/main/scala/kafka/server/KafkaServerStartable.scala Tue
Dec 13 23:55:52 2011
@@ -39,7 +39,7 @@ class KafkaServerStartable(val serverCon
     server = new KafkaServer(serverConfig)
     if (consumerConfig != null)
       embeddedConsumer =
-        new EmbeddedConsumer(consumerConfig, producerConfig, server)
+        new EmbeddedConsumer(consumerConfig, producerConfig, this)
   }
 
   def startup() {
@@ -75,7 +75,7 @@ class KafkaServerStartable(val serverCon
 
 class EmbeddedConsumer(private val consumerConfig: ConsumerConfig,
                        private val producerConfig: ProducerConfig,
-                       private val kafkaServer: KafkaServer) extends TopicEventHandler[String]
with Logging {
+                       private val kafkaServerStartable: KafkaServerStartable) extends TopicEventHandler[String]
with Logging {
 
   private val whiteListTopics =
     consumerConfig.mirrorTopicsWhitelist.split(",").toList.map(_.trim)
@@ -160,7 +160,8 @@ class EmbeddedConsumer(private val consu
   }
 
   def startup() {
-    topicEventWatcher = new ZookeeperTopicEventWatcher(consumerConfig, this)
+    info("staring up embedded consumer")
+    topicEventWatcher = new ZookeeperTopicEventWatcher(consumerConfig, this, kafkaServerStartable)
     /*
     * consumer threads are (re-)started upon topic events (which includes an
     * initial startup event which lists the current topics)



Mime
View raw message