fixing memory leak in kafka MessageHandler

2013-03-14 23:25:35 -06:00 · 2013-03-14 23:25:35 -06:00 · d069283211
parent cfa8e769a8
commit d069283211
1 changed files with 1 additions and 4 deletions
--- a/streaming/src/main/scala/spark/streaming/dstream/KafkaInputDStream.scala
+++ b/streaming/src/main/scala/spark/streaming/dstream/KafkaInputDStream.scala
@ -114,11 +114,8 @@ class KafkaReceiver(kafkaParams: Map[String, String],
  private class MessageHandler(stream: KafkaStream[String]) extends Runnable {
    def run() {
      logInfo("Starting MessageHandler.")
-      stream.takeWhile { msgAndMetadata =>
+      for (msgAndMetadata <- stream) {
        blockGenerator += msgAndMetadata.message
-        // Keep on handling messages
-
-        true
      }
    }
  }