Merge pull request #531 from ask/seek_to_committed_return_new_offsets

Consumer.seek_to_committed now returns mapping of committed offsets

Author: tvoinarovskyi

aiokafka/consumer/consumer.py

@@ -825,6 +825,10 @@ async def seek_to_committed(self, *partitions):
             *partitions: Optionally provide specific TopicPartitions, otherwise
                 default to all assigned partitions.
 
+        Returns:
+            dict: ``{TopicPartition: offset}`` mapping
+            of the currently committed offsets.
+
         Raises:
             IllegalStateError: If any partition is not currently assigned
             IllegalOperation: If used with ``group_id == None``
@@ -848,11 +852,14 @@ async def seek_to_committed(self, *partitions):
                 raise IllegalStateError(
                     "Partitions {} are not assigned".format(not_assigned))
 
+        committed_offsets = {}
         for tp in partitions:
             offset = await self.committed(tp)
+            committed_offsets[tp] = offset
             log.debug("Seeking to committed of partition %s %s", tp, offset)
             if offset and offset > 0:
                 self._fetcher.seek_to(tp, offset)
+        return committed_offsets
 
     async def offsets_for_times(self, timestamps):
         """