title: Kafka (Re-)Joining group stuck with more than 2 topics
body: <p>I'm developing a system that uses Kafka as messaging pub/sub tool.</p>
  
  <p>Data are generated by a scala script:</p>
  
  <pre><code>val kafkaParams = new Properties()
      kafkaParams.put("bootstrap.servers", "localhost:9092")
      kafkaParams.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
      kafkaParams.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
      kafkaParams.put("group.id", "test_luca")
  
      //kafka producer
      val producer = new KafkaProducer[String, String](kafkaParams)
  
      //Source list
      val s1 = new java.util.Timer()
      val tasks1 = new java.util.TimerTask {
          def run() = {
              val date = new java.util.Date
              val date2 = date.getTime()
              val send = ""+ date2 + ", 45.1234, 12.5432, 4.5, 3.0"
              val data = new ProducerRecord[String,String]("topic_s1", send)
              producer.send(data)
          }
      }
      s1.schedule(tasks1, 1000L, 1000L)
  
      val s2 = new java.util.Timer()
      val tasks2 = new java.util.TimerTask {
          def run() = {
              val date = new java.util.Date
              val date2 = date.getTime()
              val send = ""+ date2 + ", 1.111, 9.999, 10.4, 10.0"
              val data = new ProducerRecord[String,String]("topic_s2", send)
              producer.send(data)
          }
      }
      s2.schedule(tasks2, 2000L, 2000L)
  </code></pre>
  
  <p>I need to test the kafka performances in some particular situations. In one case I have an other script that consume data from topics "topic_s1" and "topic_s2", elaborate them and then produce new data with different topics (topic_s1b and topic_s2b). Subsequently these elaborated datum are consumed by an Apache Spark Streaming script. </p>
  
  <p>If I omit the consumer/producer script (I have only 1 Kafka producer with 2 topics and the Spark Script) everything works fine.</p>
  
  <p>If I use the full configuration (1 kafka producer with 2 topics, the "middleware" script that consume data from kafka producer, elaborate them and produce new data with new topics, 1 spark script that consume data with the new topics) the Spark Streaming script is stuck on <code>INFO AbstractCoordinator: (Re-)joining group test_luca</code> </p>
  
  <p>I'm running everything locally and I do not make modifications to the kafka and zookeeper configurations.</p>
  
  <p>Any suggestions?</p>
  
  <p>UPDATE: The spark script:</p>
  
  <pre><code>val sparkConf = new SparkConf().setAppName("SparkScript").set("spark.driver.allowMultipleContexts", "true").setMaster("local[2]")
  val sc = new SparkContext(sparkConf)
  
  val ssc = new StreamingContext(sc, Seconds(4))
  
  case class Thema(name: String, metadata: JObject)
  case class Tempo(unit: String, count: Int, metadata: JObject)
  case class Spatio(unit: String, metadata: JObject)
  case class Stt(spatial: Spatio, temporal: Tempo, thematic: Thema)
  case class Location(latitude: Double, longitude: Double, name: String)
  
  case class Data(location: Location, timestamp: Long, measurement: Int, unit: String, accuracy: Double)
  case class Sensor(sensor_name: String, start_date: String, end_date: String, data_schema: Array[String], data: Data, stt: Stt)
  
  
  case class Datas(location: Location, timestamp: Long, measurement: Int, unit: String, accuracy: Double)
  case class Sensor2(sensor_name: String, start_date: String, end_date: String, data_schema: Array[String], data: Datas, stt: Stt)
  
  
  val kafkaParams = Map[String, Object](
      "bootstrap.servers" -&gt; "localhost:9092",
      "key.deserializer" -&gt; classOf[StringDeserializer].getCanonicalName,
      "value.deserializer" -&gt; classOf[StringDeserializer].getCanonicalName,
      "group.id" -&gt; "test_luca",
      "auto.offset.reset" -&gt; "latest",
      "enable.auto.commit" -&gt; (false: java.lang.Boolean)
  )
  
  val topics1 = Array("topics1")
  val topics2 = Array("topics2")
  
  val stream = KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String](topics1, kafkaParams))
  val stream2 = KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String](topics2, kafkaParams))
  
  val s1 = stream.map(record =&gt; {
    implicit val formats = DefaultFormats
    parse(record.value).extract[Sensor]
  }
  )
  val s2 = stream2.map(record =&gt; {
    implicit val formats = DefaultFormats
    parse(record.value).extract[Sensor2]
  }
  )
  
  val f1 = s1.map { x =&gt; x.sensor_name }
  f1.print()
  val f2 = s2.map { x =&gt; x.sensor_name }
  f2.print()
  </code></pre>
  
  <p>Thanks
  Luca</p>