Optionally return OffsetAndMetadata from consumer.committed(tp)issue-1082-offset-and-metadata

3 files changed, 13 insertions, 6 deletions

diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py
index f9d96b0..5cb25f2 100644
--- a/kafka/consumer/fetcher.py
+++ b/kafka/consumer/fetcher.py
@@ -185,7 +185,7 @@ class Fetcher(six.Iterator):
                 self._subscriptions.need_offset_reset(tp)
                 self._reset_offset(tp)
             else:
-                committed = self._subscriptions.assignment[tp].committed
+                committed = self._subscriptions.assignment[tp].committed.offset
                 log.debug("Resetting offset for partition %s to the committed"
                           " offset %s", tp, committed)
                 self._subscriptions.seek(tp, committed)
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index cde956c..89a0292 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -525,7 +525,7 @@ class KafkaConsumer(six.Iterator):
             offsets = self._subscription.all_consumed_offsets()
         self._coordinator.commit_offsets_sync(offsets)
 
-    def committed(self, partition):
+    def committed(self, partition, metadata=False):
         """Get the last committed offset for the given partition.
 
         This offset will be used as the position for the consumer
@@ -537,9 +537,11 @@ class KafkaConsumer(six.Iterator):
 
         Arguments:
             partition (TopicPartition): The partition to check.
+            metadata (bool, optional): If True, return OffsetAndMetadata struct
+                instead of offset int. Default: False.
 
         Returns:
-            The last committed offset, or None if there was no prior commit.
+            The last committed offset (int or OffsetAndMetadata), or None if there was no prior commit.
         """
         assert self.config['api_version'] >= (0, 8, 1), 'Requires >= Kafka 0.8.1'
         assert self.config['group_id'] is not None, 'Requires group_id'
@@ -553,10 +555,15 @@ class KafkaConsumer(six.Iterator):
         else:
             commit_map = self._coordinator.fetch_committed_offsets([partition])
             if partition in commit_map:
-                committed = commit_map[partition].offset
+                committed = commit_map[partition]
             else:
                 committed = None
-        return committed
+
+        if committed is not None:
+            if metadata:
+                return committed
+            else:
+                return committed.offset
 
     def _fetch_all_topic_metadata(self):
         """A blocking call that fetches topic metadata for all topics in the
diff --git a/kafka/consumer/subscription_state.py b/kafka/consumer/subscription_state.py
index 76a6c50..08842d1 100644
--- a/kafka/consumer/subscription_state.py
+++ b/kafka/consumer/subscription_state.py
@@ -374,7 +374,7 @@ class SubscriptionState(object):
 
 class TopicPartitionState(object):
     def __init__(self):
-        self.committed = None # last committed position
+        self.committed = None # last committed OffsetAndMetadata
         self.has_valid_position = False # whether we have valid position
         self.paused = False # whether this partition has been paused by the user
         self.awaiting_reset = False # whether we are awaiting reset