Merge 832a5de758 into fb54d8b549

Signed-off-by: Rajiv Ranjan Singh <rajivperfect007@gmail.com>

src/scala/com/twitter/recos/hose/common/UnifiedGraphWriter.scala

@@ -85,16 +85,13 @@ trait UnifiedGraphWriter[
     }
   }
 
-  private def initRecosHoseKafka(
-    queue: java.util.Queue[Array[RecosHoseMessage]],
-    queuelimit: Semaphore,
-  ): Unit = {
+private def initRecosHoseKafka(queue: java.util.Queue[Array[RecosHoseMessage]], queuelimit: Semaphore): Unit = {
   try {
-      consumers = (0 until consumerNum).map { index =>
-        new ThreadSafeKafkaConsumerClient(
-          kafkaConsumerBuilder.clientId(s"clientId-$index").enableAutoCommit(false).config)
+    val consumers = (0 until consumerNum).map { index =>
+      new ThreadSafeKafkaConsumerClient(kafkaConsumerBuilder.clientId(s"clientId-$index").enableAutoCommit(false).config)
     }
-      processors = consumers.zipWithIndex.map {
+
+    val processors = consumers.zipWithIndex.map {
       case (consumer, index) =>
         val bufferedWriter = BufferedEdgeCollector(bufferSize, queue, queuelimit, statsReceiver)
         val processor = RecosEdgeProcessor(bufferedWriter)(statsReceiver)
@@ -112,21 +109,14 @@ trait UnifiedGraphWriter[
     }
 
     log.info(s"starting ${processors.size} recosKafka processors")
-      processors.foreach { processor =>
-        processor.start()
-      }
+    processors.foreach(_.start())
+
   } catch {
     case e: Throwable =>
       e.printStackTrace()
       log.error(e, e.toString)
-        processors.foreach { processor =>
-          processor.close()
-        }
-        processors = Seq.empty
-        consumers.foreach { consumer =>
-          consumer.close()
-        }
-        consumers = Seq.empty
+      processors.foreach(_.close())
+      consumers.foreach(_.close())
   }
 }