code: def _send_commit_request(self, retry_delay=None, attempt=None):
          
          # If there's a _commit_call, and it's not active, clear it, it probably
          # just called us...
          if self._commit_call and not self._commit_call.active():
              self._commit_call = None
  
          # Make sure we only have one outstanding commit request at a time
          if self._commit_req is not None:
              raise OperationInProgress(self._commit_req)
  
          # Handle defaults
          if retry_delay is None:
              retry_delay = self.retry_init_delay
          if attempt is None:
              attempt = 1
  
          # Create new OffsetCommitRequest with the latest processed offset
          commit_offset = self._last_processed_offset
          commit_request = OffsetCommitRequest(
              self.topic, self.partition, commit_offset,
              TIMESTAMP_INVALID, self.commit_metadata)
          log.debug("Committing off=%d grp=%s tpc=%s part=%s req=%r",
                    self._last_processed_offset, self.consumer_group,
                    self.topic, self.partition, commit_request)
  
          # Send the request, add our callbacks
          self._commit_req = d = self.client.send_offset_commit_request(
              self.consumer_group, [commit_request])
  
          d.addBoth(self._clear_commit_req)
          d.addCallbacks(
              self._update_committed_offset, self._handle_commit_error,
              callbackArgs=(commit_offset,),
              errbackArgs=(retry_delay, attempt))
docs: Send a commit request with our last_processed_offset