title: CommitFailedException Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member
body: <p>I was using kafka 0.10.2 and now faced a CommitFailedException. like:</p>
  
  <blockquote>
    <p>Commit cannot be completed since the group has already rebalanced and
    assigned the partitions to another member. This means that the time
    between subsequent calls to poll() was longer than the configured
    max.poll.interval.ms, which typically implies that the poll loop is
    spending too much time message processing. You can address this either
    by increasing the session timeout or by reducing the maximum size of
    batches returned in poll() with max.poll.records.</p>
  </blockquote>
  
  <p>I have set max.poll.interval.ms to Integer.MAX_VALUE. so can anyone tell me why this still happens even I have set the value ? </p>
  
  <p>Another question is:
  I do as description to set session.timeout.ms to 60000 and it still happens. I try to reproduce by a simple code</p>
  
  <pre><code> public static void main(String[] args) throws InterruptedException {     
          Logger logger = Logger.getLogger(KafkaConsumer10.class);
          logger.info("XX");
          Properties props = new Properties();
          props.put("bootstrap.servers", "kafka-broker:9098");
          props.put("group.id", "test");
          props.put("enable.auto.commit", "true");
          props.put("auto.commit.interval.ms", "1000");
          props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
          props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
          props.put("max.poll.interval.ms", "300000");
          props.put("session.timeout.ms", "10000");
          props.put("max.poll.records", "2");
          KafkaConsumer&lt;String, String&gt; consumer = new KafkaConsumer&lt;&gt;(props);
          consumer.subscribe(Arrays.asList("t1"));
          while (true) {
              Thread.sleep(11000);
              ConsumerRecords&lt;String, String&gt; records = consumer.poll(100);
              //Thread.sleep(11000);
              Thread.sleep(11000);
              for (ConsumerRecord&lt;String, String&gt; record : records)
                  System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
          }
  </code></pre>
  
  <p>when I set session.timeout.ms to 10000, I try to sleep more than 10000 ms in my poll loop, but it seems work and no Exception out. so I'm confused about this. if heartbeat is triggered by consumer.poll and consumer.commit, seems heartbeat is out of session timeout in my code. why not throw CommitFailedException ? </p>