title: kafka Python - CommitFailedError
body: <p><strong>Problem:</strong> when calling <code>commit()</code>, I am getting the error if message execution is taking a long time. </p>
  
  <p>Here is the code I am using :</p>
  
  <pre><code>consumer = KafkaConsumer(topic,
        group_id='my-group',
        value_deserializer=lambda m: json.loads(m.decode('utf-8')),
        bootstrap_servers=kafka_server_url,
        api_version=(0, 10, 1),
        auto_offset_reset='earliest',
        enable_auto_commit= False,
        #consumer_timeout_ms=3000,
        session_timeout_ms=250000, #The timeout used to detect failures when using Kafka group management facilities. 
        request_timeout_ms=300000,
        heartbeat_interval_ms=80000,
        #max_poll_records=100,#The maximum number of records returned in a single call to poll(). Default: 500
        enable_auto_commit=False,
        retry_backoff_ms = 1000
        )
  errored = False
  while errored is False:
    logging.warn("Polling new messages from queue ......")
    messages = []
    crs = [] # Store all consumer records
    try:
      tpd = (consumer.poll(max_records=1))
      [crs.extend(tp) for tp in tpd.values()] # List of cr's
      [messages.extend([cr.value]) for cr in crs] 
      if(len(messages)):
        consumer.commit()
        logging.warn("{} messages available in queue".format(len(messages)))
        for msg in messages:
          if(self.__validators.is_valid_message(msg)):
            self.process_message(msg);
          else:
            continue
        # If auto_commit_enable is False, remember to commit() periodically    
        logging.warn("All messages processed. Commiting kafka consumer")
  
    except Exception as e:
      message = "Error occured while processing message Exception {}".format(e)
      logging.critical(message,exc_info=True)
      errored = True
      break
    logging.info("Sleeping for 5 Seconds")
    time.sleep(3)
  
  def process_message(self,message):
   print("Processed")
  </code></pre>
  
  <p>Following is the error trace:</p>
  
  <blockquote>
    <p>Traceback (most recent call last): File "message_consumer.py", line
    413, in consume_messages consumer.commit() File
    "/usr/lib/python3.6/site-packages/kafka/consumer/group.py", line 472,
    in commit self._coordinator.commit_offsets_sync(offsets) File
    "/usr/lib/python3.6/site-packages/kafka/coordinator/consumer.py", line
    398, in commit_offsets_sync raise future.exception # pylint:
    disable-msg=raising-bad-type kafka.errors.CommitFailedError:
    CommitFailedError: 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 session.timeout.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>My message is not getting committed and it is getting reprocessed from kafka queue. How do I solve this? Please help.</p>
  
  <p>Any help will be appreciated. Thanks in advance.</p>