Support consumer metadata requests

Support added for ConsumerMetadataRequest and ConsumerMetadataResponse
Added consumer-aware request routine for supporting the consumer coordinator
Added separate client method for fetching Kafka-committed offsets from the coordinator

1 files changed, 22 insertions, 0 deletions

diff --git a/kafka/common.py b/kafka/common.py
index 66987ff..a7d8164 100644
--- a/kafka/common.py
+++ b/kafka/common.py
@@ -13,6 +13,13 @@ MetadataRequest = namedtuple("MetadataRequest",
 MetadataResponse = namedtuple("MetadataResponse",
     ["brokers", "topics"])
 
+# https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ConsumerMetadataRequest
+ConsumerMetadataRequest = namedtuple("ConsumerMetadataRequest",
+    ["groups"])
+
+ConsumerMetadataResponse = namedtuple("ConsumerMetadataResponse",
+    ["error", "nodeId", "host", "port"])
+
 # https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ProduceAPI
 ProduceRequest = namedtuple("ProduceRequest",
     ["topic", "partition", "messages"])
@@ -160,6 +167,21 @@ class StaleLeaderEpochCodeError(BrokerResponseError):
     message = 'STALE_LEADER_EPOCH_CODE'
 
 
+class OffsetsLoadInProgressCode(BrokerResponseError):
+    errno = 14
+    message = 'OFFSETS_LOAD_IN_PROGRESS_CODE'
+
+
+class ConsumerCoordinatorNotAvailableCode(BrokerResponseError):
+    errno = 15
+    message = 'CONSUMER_COORDINATOR_NOT_AVAILABLE_CODE'
+
+
+class NotCoordinatorForConsumerCode(BrokerResponseError):
+    errno = 16
+    message = 'NOT_COORDINATOR_FOR_CONSUMER_CODE'
+
+
 class KafkaUnavailableError(KafkaError):
     pass