title: No current assignment for partition: <topic-partition> during re-joining; adding new partitions during run-time doesnt work
body: <p>I have a streaming application that uses <a href="https://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html" rel="nofollow noreferrer">Kafka+SparkStreaming</a>. I start with one partition, one consumer instance, three brokers.
  I want to do following two things in sequence:</p>
  
  <p>1) Add new partitions to a topic during runtime, which I do using:</p>
  
  <pre><code>bin/kafka-topics.sh --zookeeper 192.168.101.164:2181 --alter --topic topic10 --partitions 2
  </code></pre>
  
  <p>2) Add new consumer instances to a particular consumer group which I do by running the same consumer code as a separate process.</p>
  
  <p>However, on doing 1:  I do not see the (only) consumer instance consuming the newly added partition. It only consumes the partitions which it was initially assigned automatically by the zookeeper.</p>
  
  <p>On doing 2: a)The new instance is automatically assigned one partition <code>topic10-1</code> which happens successfully but b)The first instance rejoins after rebalancing but fails with the following error inspite of a)</p>
  
  <pre><code>17/06/07 16:47:02 INFO ConsumerCoordinator: Revoking previously assigned partitions [topic10-0] for group SparkConsumerGrp
  17/06/07 16:47:02 INFO AbstractCoordinator: (Re-)joining group SparkConsumerGrp
  17/06/07 16:47:02 INFO AbstractCoordinator: Successfully joined group SparkConsumerGrp with generation 17
  17/06/07 16:47:02 INFO ConsumerCoordinator: Setting newly assigned partitions [topic10-0] for group SparkConsumerGrp
  17/06/07 16:47:02 ERROR JobScheduler: Error generating jobs for time 1496879222000 ms
  java.lang.IllegalStateException: No current assignment for partition topic10-1
      at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedState(SubscriptionState.java:264)
      at org.apache.kafka.clients.consumer.internals.SubscriptionState.needOffsetReset(SubscriptionState.java:336)
      at org.apache.kafka.clients.consumer.KafkaConsumer.seekToEnd(KafkaConsumer.java:1236)
      at org.apache.spark.streaming.kafka010.DirectKafkaInputDStream.latestOffsets(DirectKafkaInputDStream.scala:197)
      at org.apache.spark.streaming.kafka010.DirectKafkaInputDStream.compute(DirectKafkaInputDStream.scala:214)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
      at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
      at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:335)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:333)
      at scala.Option.orElse(Option.scala:289)
      at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:330)
      at org.apache.spark.streaming.dstream.MappedDStream.compute(MappedDStream.scala:36)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
      at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
      at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:335)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:333)
      at scala.Option.orElse(Option.scala:289)
      at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:330)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2$$anonfun$apply$29.apply(DStream.scala:900)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2$$anonfun$apply$29.apply(DStream.scala:899)
      at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
      at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
      at scala.collection.Iterator$class.foreach(Iterator.scala:893)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
      at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
      at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
      at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
      at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2.apply(DStream.scala:899)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2.apply(DStream.scala:877)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
      at org.apache.spark.SparkContext.withScope(SparkContext.scala:701)
      at org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264)
      at org.apache.spark.streaming.dstream.DStream.slice(DStream.scala:877)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$1.apply(DStream.scala:871)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$1.apply(DStream.scala:871)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
      at org.apache.spark.SparkContext.withScope(SparkContext.scala:701)
      at org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264)
      at org.apache.spark.streaming.dstream.DStream.slice(DStream.scala:870)
      at org.apache.spark.streaming.dstream.WindowedDStream.compute(WindowedDStream.scala:65)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
      at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
      at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:335)
      at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:333)
      at scala.Option.orElse(Option.scala:289)
      at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:330)
      at org.apache.spark.streaming.dstream.ForEachDStream.generateJob(ForEachDStream.scala:48)
      at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:117)
      at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:116)
      at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
      at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
      at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
      at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
      at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
      at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
      at org.apache.spark.streaming.DStreamGraph.generateJobs(DStreamGraph.scala:116)
      at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$3.apply(JobGenerator.scala:249)
      at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$3.apply(JobGenerator.scala:247)
      at scala.util.Try$.apply(Try.scala:192)
      at org.apache.spark.streaming.scheduler.JobGenerator.generateJobs(JobGenerator.scala:247)
      at org.apache.spark.streaming.scheduler.JobGenerator.org$apache$spark$streaming$scheduler$JobGenerator$$processEvent(JobGenerator.scala:183)
      at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:89)
      at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:88)
      at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
  </code></pre>
  
  <p>I notice that the time at which the new instance successfully joins the group exactly matches the time that the first one fails every time.
  I am using <code>Kafka version 0.10.0, Kafka-client version 0.10.0.1, spark-streaming-kafka-0-10_2.11 version 2.1.1</code>.</p>
  
  <p>Finally, here's the consumer code:</p>
  
  <pre><code>    SparkConf sparkConf = new  SparkConf().setMaster("local[5]").setAppName("SparkConsumer1").set("spark.driver.host", "localhost");
  
      JavaSparkContext sc = new JavaSparkContext(sparkConf);
  
      // Create a StreamingContext with a 1 second batch size
      JavaStreamingContext jssc = new JavaStreamingContext(sc, Durations.seconds(15));    
      List&lt;String&gt; topicSet = Arrays.asList(topics.split(","));
  
      Map&lt;String, Object&gt; kafkaParams = new HashMap&lt;&gt;();
      kafkaParams.put("bootstrap.servers", brokers);
      kafkaParams.put("auto.offset.reset", "latest");
      kafkaParams.put("group.id", "SparkConsumerGrp3");
      kafkaParams.put("key.deserializer", StringDeserializer.class);
      kafkaParams.put("value.deserializer", StringDeserializer.class);
      kafkaParams.put("zookeeper.connect", "192.168.101.164:2181");
      kafkaParams.put("enable.auto.commit", "true");
      kafkaParams.put("auto.commit.interval.ms", "1000");
      kafkaParams.put("session.timeout.ms","30000");
  
      final JavaInputDStream&lt;ConsumerRecord&lt;String, String&gt;&gt; messages =
                KafkaUtils.createDirectStream(
                  jssc,
                  LocationStrategies.PreferBrokers(),
                  ConsumerStrategies.&lt;String, String&gt;Subscribe(topicSet, kafkaParams)
                );
  </code></pre>
  
  <p>I also found something similar <a href="https://issues.apache.org/jira/browse/SPARK-19547" rel="nofollow noreferrer">here</a>, but there was no real answer. I would really really appreciate any help. Thanks.</p>
  
  <p><strong>Update</strong>
  For 1) I  was able to have the consumer fetch data from newly added partitions. We can configure how often we want the metadata to be refreshed using <code>metadata.max.age.ms</code> which is set to <code>300000</code> by default. </p>