code: def commit_offsets_async(self, offsets, callback=None):
          
          self._invoke_completed_offset_commit_callbacks()
          if not self.coordinator_unknown():
              future = self._do_commit_offsets_async(offsets, callback)
          else:
              # we don't know the current coordinator, so try to find it and then
              # send the commit or fail (we don't want recursive retries which can
              # cause offset commits to arrive out of order). Note that there may
              # be multiple offset commits chained to the same coordinator lookup
              # request. This is fine because the listeners will be invoked in the
              # same order that they were added. Note also that BaseCoordinator
              # prevents multiple concurrent coordinator lookup requests.
              future = self.lookup_coordinator()
              future.add_callback(lambda r: functools.partial(self._do_commit_offsets_async, offsets, callback)())
              if callback:
                  future.add_errback(lambda e: self.completed_offset_commits.appendleft((callback, offsets, e)))
  
          # ensure the commit has a chance to be transmitted (without blocking on
          # its completion). Note that commits are treated as heartbeats by the
          # coordinator, so there is no need to explicitly allow heartbeats
          # through delayed task execution.
          self._client.poll(timeout_ms=0) # no wakeup if we add that feature
  
          return future
docs: Commit specific offsets asynchronously.
  
          Arguments:
              offsets (dict {TopicPartition: OffsetAndMetadata}): what to commit
              callback (callable, optional): called as callback(offsets, response)
                  response will be either an Exception or a OffsetCommitResponse
                  struct. This callback can be used to trigger custom actions when
                  a commit request completes.
  
          Returns:
              kafka.future.Future