title: kafka consumer group is rebalancing
body: <p>I am using Kafka .9 and new java consumer. I am polling inside a loop. I am getting commitfailedexcption because of group rebalance, when code try to execute consumer.commitSycn . Please note, I am adding session.timeout.ms as 30000 and heartbeat.interval.ms as 10000 to consumer and polling happens for sure in 30000. Can anyone help me out. Please let me know if any information is needed.</p>
  
  <p>Here is the code :-</p>
  
  <pre><code>       Properties props = new Properties();
          props.put("bootstrap.servers", {allthreeservers});
          props.put("group.id", groupId);
          props.put("key.deserializer", StringDeserializer.class.getName());
          props.put("value.deserializer", ObjectSerializer.class.getName());
          props.put("auto.offset.reset", erlierst);
          props.put("enable.auto.commit", false);
          props.put("session.timeout.ms", 30000);
          props.put("heartbeat.interval.ms", 10000);
          props.put("request.timeout.ms", 31000);
          props.put("kafka.consumer.topic.name", topic);
          props.put("max.partition.fetch.bytes", 1000);   
  
          while (true) {
              Boolean isPassed = true;
              try {
                  ConsumerRecords&lt;Object, Object&gt; records = consumer.poll(1000);
                  if (records.count() &gt; 0) {
                          ConsumeEventInThread consumerEventInThread = new ConsumeEventInThread(records, consumerService);
                          FutureTask&lt;Boolean&gt; futureTask = new FutureTask&lt;&gt;(consumerEventInThread);
                          executorServiceForAsyncKafkaEventProcessing.execute(futureTask);
                          try {
                              isPassed = (Boolean) futureTask.get(Long.parseLong(props.getProperty("session.timeout.ms")) - Long.parseLong("5000"), TimeUnit.MILLISECONDS);
                          } catch (Exception Exception) {
                              logger.warn("Time out after waiting till session time out");
                          }
                          consumer.commitSync();
                          logger.info("Successfully committed offset for topic " + Arrays.asList(props.getProperty("kafka.consumer.topic.name")));
                      }else{
                          logger.info("Failed to process consumed messages, will not Commit and consume again");
                      }
  
                  }
  
              } catch (Exception e) {
                  logger.error("Unhandled exception in while consuming messages " + Arrays.asList(props.getProperty("kafka.consumer.topic.name")), e);
              }
          }
  </code></pre>