Repository: dpkp/kafka-python
Pull request #2503 (2025-02-27 01:56:17)

Support 2.1 baseline consumer group apis
Minor updates to JoinGroup/SyncGroup/LeaveGroup/Heartbeat to support throttle_time_ms field.


Patch:
diff --git a/kafka/admin/client.py b/kafka/admin/client.py
index 68b0af1..a46cf9c 100644
--- a/kafka/admin/client.py
+++ b/kafka/admin/client.py
@@ -23,7 +23,8 @@ from kafka.protocol.admin import (
     ListGroupsRequest, DescribeGroupsRequest, DescribeAclsRequest, CreateAclsRequest, DeleteAclsRequest,
     DeleteGroupsRequest, DescribeLogDirsRequest
 )
-from kafka.protocol.commit import GroupCoordinatorRequest, OffsetFetchRequest
+from kafka.protocol.commit import OffsetFetchRequest
+from kafka.protocol.find_coordinator import FindCoordinatorRequest
 from kafka.protocol.metadata import MetadataRequest
 from kafka.protocol.types import Array
 from kafka.structs import TopicPartition, OffsetAndMetadata, MemberInformation, GroupInformation
@@ -285,17 +286,14 @@ class KafkaAdminClient(object):
         Returns:
             A message future
         """
-        # TODO add support for dynamically picking version of
-        # GroupCoordinatorRequest which was renamed to FindCoordinatorRequest.
-        # When I experimented with this, the coordinator value returned in
-        # GroupCoordinatorResponse_v1 didn't match the value returned by
-        # GroupCoordinatorResponse_v0 and I couldn't figure out why.
-        version = self._client.api_version(GroupCoordinatorRequest, max_version=0)
+        version = self._client.api_version(FindCoordinatorRequest, max_version=2)
         if version <= 0:
-            request = GroupCoordinatorRequest[version](group_id)
+            request = FindCoordinatorRequest[version](group_id)
+        elif version <= 2:
+            request = FindCoordinatorRequest[version](group_id, 0)
         else:
             raise NotImplementedError(
-                "Support for GroupCoordinatorRequest_v{} has not yet been added to KafkaAdminClient."
+                "Support for FindCoordinatorRequest_v{} has not yet been added to KafkaAdminClient."
                 .format(version))
         return self._send_request_to_node(self._client.least_loaded_node(), request)
 
@@ -308,18 +306,13 @@ class KafkaAdminClient(object):
         Returns:
             The node_id of the broker that is the coordinator.
         """
-        if response.API_VERSION <= 0:
-            error_type = Errors.for_code(response.error_code)
-            if error_type is not Errors.NoError:
-                # Note: When error_type.retriable, Java will retry... see
-                # KafkaAdminClient's handleFindCoordinatorError method
-                raise error_type(
-                    "FindCoordinatorRequest failed with response '{}'."
-                    .format(response))
-        else:
-            raise NotImplementedError(
-                "Support for FindCoordinatorRequest_v{} has not yet been added to KafkaAdminClient."
-                .format(response.API_VERSION))
+        error_type = Errors.for_code(response.error_code)
+        if error_type is not Errors.NoError:
+            # Note: When error_type.retriable, Java will retry... see
+            # KafkaAdminClient's handleFindCoordinatorError method
+            raise error_type(
+                "FindCoordinatorRequest failed with response '{}'."
+                .format(response))
         return response.coordinator_id
 
     def _find_coordinator_ids(self, group_ids):
diff --git a/kafka/cluster.py b/kafka/cluster.py
index 69a49de..98272ea 100644
--- a/kafka/cluster.py
+++ b/kafka/cluster.py
@@ -21,7 +21,7 @@ class ClusterMetadata(object):
     A class to manage kafka cluster metadata.
 
     This class does not perform any IO. It simply updates internal state
-    given API responses (MetadataResponse, GroupCoordinatorResponse).
+    given API responses (MetadataResponse, FindCoordinatorResponse).
 
     Keyword Arguments:
         retry_backoff_ms (int): Milliseconds to backoff when retrying on
@@ -367,8 +367,8 @@ class ClusterMetadata(object):
         """Update with metadata for a group coordinator
 
         Arguments:
-            group (str): name of group from GroupCoordinatorRequest
-            response (GroupCoordinatorResponse): broker response
+            group (str): name of group from FindCoordinatorRequest
+            response (FindCoordinatorResponse): broker response
 
         Returns:
             string: coordinator node_id if metadata is updated, None on error
@@ -376,7 +376,7 @@ class ClusterMetadata(object):
         log.debug("Updating coordinator for %s: %s", group, response)
         error_type = Errors.for_code(response.error_code)
         if error_type is not Errors.NoError:
-            log.error("GroupCoordinatorResponse error: %s", error_type)
+            log.error("FindCoordinatorResponse error: %s", error_type)
             self._groups[group] = -1
             return
 
diff --git a/kafka/conn.py b/kafka/conn.py
index 347e500..4065d1c 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -26,7 +26,7 @@ from kafka.metrics.stats import Avg, Count, Max, Rate
 from kafka.oauth.abstract import AbstractTokenProvider
 from kafka.protocol.admin import SaslHandShakeRequest, DescribeAclsRequest, DescribeClientQuotasRequest
 from kafka.protocol.commit import OffsetFetchRequest
-from kafka.protocol.offset import OffsetRequest
+from kafka.protocol.list_offsets import ListOffsetsRequest
 from kafka.protocol.produce import ProduceRequest
 from kafka.protocol.metadata import MetadataRequest
 from kafka.protocol.fetch import FetchRequest
@@ -1202,7 +1202,7 @@ class BrokerConnection(object):
             ((2, 5), DescribeAclsRequest[2]),
             ((2, 4), ProduceRequest[8]),
             ((2, 3), FetchRequest[11]),
-            ((2, 2), OffsetRequest[5]),
+            ((2, 2), ListOffsetsRequest[5]),
             ((2, 1), FetchRequest[10]),
             ((2, 0), FetchRequest[8]),
             ((1, 1), FetchRequest[7]),
@@ -1255,14 +1255,15 @@ class BrokerConnection(object):
         from kafka.protocol.admin import ListGroupsRequest
         from kafka.protocol.api_versions import ApiVersionsRequest
         from kafka.protocol.broker_api_versions import BROKER_API_VERSIONS
-        from kafka.protocol.commit import OffsetFetchRequest, GroupCoordinatorRequest
+        from kafka.protocol.commit import OffsetFetchRequest
+        from kafka.protocol.find_coordinator import FindCoordinatorRequest
 
         test_cases = [
             # All cases starting from 0.10 will be based on ApiVersionsResponse
             ((0, 11), ApiVersionsRequest[1]()),
             ((0, 10, 0), ApiVersionsRequest[0]()),
             ((0, 9), ListGroupsRequest[0]()),
-            ((0, 8, 2), GroupCoordinatorRequest[0]('kafka-python-default-group')),
+            ((0, 8, 2), FindCoordinatorRequest[0]('kafka-python-default-group')),
             ((0, 8, 1), OffsetFetchRequest[0]('kafka-python-default-group', [])),
             ((0, 8, 0), MetadataRequest[0](topics)),
         ]
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py
index 9dd4b84..c6886c4 100644
--- a/kafka/consumer/fetcher.py
+++ b/kafka/consumer/fetcher.py
@@ -13,8 +13,8 @@ import kafka.errors as Errors
 from kafka.future import Future
 from kafka.metrics.stats import Avg, Count, Max, Rate
 from kafka.protocol.fetch import FetchRequest
-from kafka.protocol.offset import (
-    OffsetRequest, OffsetResetStrategy, UNKNOWN_OFFSET
+from kafka.protocol.list_offsets import (
+    ListOffsetsRequest, OffsetResetStrategy, UNKNOWN_OFFSET
 )
 from kafka.record import MemoryRecords
 from kafka.serializer import Deserializer
@@ -272,7 +272,7 @@ class Fetcher(six.Iterator):
             if not timestamps:
                 return {}
 
-            future = self._send_offset_requests(timestamps)
+            future = self._send_list_offsets_requests(timestamps)
             self._client.poll(future=future, timeout_ms=remaining_ms)
 
             if future.succeeded():
@@ -411,10 +411,10 @@ class Fetcher(six.Iterator):
 
             tp = self._next_partition_records.topic_partition
 
-            # We can ignore any prior signal to drop pending message sets
+            # We can ignore any prior signal to drop pending record batches
             # because we are starting from a fresh one where fetch_offset == position
             # i.e., the user seek()'d to this position
-            self._subscriptions.assignment[tp].drop_pending_message_set = False
+            self._subscriptions.assignment[tp].drop_pending_record_batch = False
 
             for msg in self._next_partition_records.take():
 
@@ -430,12 +430,12 @@ class Fetcher(six.Iterator):
                     break
 
                 # If there is a seek during message iteration,
-                # we should stop unpacking this message set and
+                # we should stop unpacking this record batch and
                 # wait for a new fetch response that aligns with the
                 # new seek position
-                elif self._subscriptions.assignment[tp].drop_pending_message_set:
-                    log.debug("Skipping remainder of message set for partition %s", tp)
-                    self._subscriptions.assignment[tp].drop_pending_message_set = False
+                elif self._subscriptions.assignment[tp].drop_pending_record_batch:
+                    log.debug("Skipping remainder of record batch for partition %s", tp)
+                    self._subscriptions.assignment[tp].drop_pending_record_batch = False
                     self._next_partition_records = None
                     break
 
@@ -451,7 +451,7 @@ class Fetcher(six.Iterator):
 
             self._next_partition_records = None
 
-    def _unpack_message_set(self, tp, records):
+    def _unpack_records(self, tp, records):
         try:
             batch = records.next_batch()
             while batch is not None:
@@ -459,8 +459,8 @@ class Fetcher(six.Iterator):
                 # Try DefaultsRecordBatch / message log format v2
                 # base_offset, last_offset_delta, and control batches
                 try:
-                    self._subscriptions.assignment[tp].last_offset_from_message_batch = batch.base_offset + \
-                                                                                        batch.last_offset_delta
+                    self._subscriptions.assignment[tp].last_offset_from_record_batch = batch.base_offset + \
+                                                                                       batch.last_offset_delta
                     # Control batches have a single record indicating whether a transaction
                     # was aborted or committed.
                     # When isolation_level is READ_COMMITTED (currently unsupported)
@@ -519,7 +519,7 @@ class Fetcher(six.Iterator):
             return f.deserialize(topic, bytes_)
         return f(bytes_)
 
-    def _send_offset_requests(self, timestamps):
+    def _send_list_offsets_requests(self, timestamps):
         """Fetch offsets for each partition in timestamps dict. This may send
         request to multiple nodes, based on who is Leader for partition.
 
@@ -564,13 +564,13 @@ class Fetcher(six.Iterator):
                 list_offsets_future.failure(err)
 
         for node_id, timestamps in six.iteritems(timestamps_by_node):
-            _f = self._send_offset_request(node_id, timestamps)
+            _f = self._send_list_offsets_request(node_id, timestamps)
             _f.add_callback(on_success)
             _f.add_errback(on_fail)
         return list_offsets_future
 
-    def _send_offset_request(self, node_id, timestamps):
-        version = self._client.api_version(OffsetRequest, max_version=1)
+    def _send_list_offsets_request(self, node_id, timestamps):
+        version = self._client.api_version(ListOffsetsRequest, max_version=3)
         by_topic = collections.defaultdict(list)
         for tp, timestamp in six.iteritems(timestamps):
             if version >= 1:
@@ -579,7 +579,16 @@ class Fetcher(six.Iterator):
                 data = (tp.partition, timestamp, 1)
             by_topic[tp.topic].append(data)
 
-        request = OffsetRequest[version](-1, list(six.iteritems(by_topic)))
+        if version <= 1:
+            request = ListOffsetsRequest[version](
+                    -1,
+                    list(six.iteritems(by_topic)))
+        else:
+            request = ListOffsetsRequest[version](
+                    -1,
+                    self._isolation_level,
+                    list(six.iteritems(by_topic)))
+
 
         # Client returns a future that only fails on network issues
         # so create a separate future and attach a callback to update it
@@ -587,20 +596,22 @@ class Fetcher(six.Iterator):
         future = Future()
 
         _f = self._client.send(node_id, request)
-        _f.add_callback(self._handle_offset_response, future)
+        _f.add_callback(self._handle_list_offsets_response, future)
         _f.add_errback(lambda e: future.failure(e))
         return future
 
-    def _handle_offset_response(self, future, response):
-        """Callback for the response of the list offset call above.
+    def _handle_list_offsets_response(self, future, response):
+        """Callback for the response of the ListOffsets api call
 
         Arguments:
             future (Future): the future to update based on response
-            response (OffsetResponse): response from the server
+            response (ListOffsetsResponse): response from the server
 
         Raises:
             AssertionError: if response does not match partition
         """
+        if response.API_VERSION >= 2 and response.throttle_time_ms > 0:
+            log.warning("ListOffsetsRequest throttled by broker (%d ms)", response.throttle_time_ms)
         timestamp_offset_map = {}
         for topic, part_data in response.topics:
             for partition_info in part_data:
@@ -610,18 +621,18 @@ class Fetcher(six.Iterator):
                 if error_type is Errors.NoError:
                     if response.API_VERSION == 0:
                         offsets = partition_info[2]
-                        assert len(offsets) <= 1, 'Expected OffsetResponse with one offset'
+                        assert len(offsets) <= 1, 'Expected ListOffsetsResponse with one offset'
                         if not offsets:
                             offset = UNKNOWN_OFFSET
                         else:
                             offset = offsets[0]
-                        log.debug("Handling v0 ListOffsetResponse response for %s. "
+                        log.debug("Handling v0 ListOffsetsResponse response for %s. "
                                   "Fetched offset %s", partition, offset)
                         if offset != UNKNOWN_OFFSET:
                             timestamp_offset_map[partition] = (offset, None)
                     else:
                         timestamp, offset = partition_info[2:]
-                        log.debug("Handling ListOffsetResponse response for %s. "
+                        log.debug("Handling ListOffsetsResponse response for %s. "
                                   "Fetched offset %s, timestamp %s",
                                   partition, offset, timestamp)
                         if offset != UNKNOWN_OFFSET:
@@ -638,7 +649,7 @@ class Fetcher(six.Iterator):
                     future.failure(error_type(partition))
                     return
                 elif error_type is Errors.UnknownTopicOrPartitionError:
-                    log.warning("Received unknown topic or partition error in ListOffset "
+                    log.warning("Received unknown topic or partition error in ListOffsets "
                              "request for partition %s. The topic/partition " +
                              "may not exist or the user may not have Describe access "
                              "to it.", partition)
@@ -673,17 +684,18 @@ class Fetcher(six.Iterator):
         """
         # create the fetch info as a dict of lists of partition info tuples
         # which can be passed to FetchRequest() via .items()
+        version = self._client.api_version(FetchRequest, max_version=6)
         fetchable = collections.defaultdict(lambda: collections.defaultdict(list))
 
         for partition in self._fetchable_partitions():
             node_id = self._client.cluster.leader_for_partition(partition)
 
             # advance position for any deleted compacted messages if required
-            if self._subscriptions.assignment[partition].last_offset_from_message_batch:
-                next_offset_from_batch_header = self._subscriptions.assignment[partition].last_offset_from_message_batch + 1
+            if self._subscriptions.assignment[partition].last_offset_from_record_batch:
+                next_offset_from_batch_header = self._subscriptions.assignment[partition].last_offset_from_record_batch + 1
                 if next_offset_from_batch_header > self._subscriptions.assignment[partition].position:
                     log.debug(
-                        "Advance position for partition %s from %s to %s (last message batch location plus one)"
+                        "Advance position for partition %s from %s to %s (last record batch location plus one)"
                         " to correct for deleted compacted messages and/or transactional control records",
                         partition, self._subscriptions.assignment[partition].position, next_offset_from_batch_header)
                     self._subscriptions.assignment[partition].position = next_offset_from_batch_header
@@ -697,11 +709,19 @@ class Fetcher(six.Iterator):
                 self._client.cluster.request_update()
 
             elif self._client.in_flight_request_count(node_id) == 0:
-                partition_info = (
-                    partition.partition,
-                    position,
-                    self.config['max_partition_fetch_bytes']
-                )
+                if version < 5:
+                    partition_info = (
+                        partition.partition,
+                        position,
+                        self.config['max_partition_fetch_bytes']
+                    )
+                else:
+                    partition_info = (
+                        partition.partition,
+                        position,
+                        -1, # log_start_offset is used internally by brokers / replicas only
+                        self.config['max_partition_fetch_bytes'],
+                    )
                 fetchable[node_id][partition.topic].append(partition_info)
                 log.debug("Adding fetch request for partition %s at offset %d",
                           partition, position)
@@ -709,40 +729,40 @@ class Fetcher(six.Iterator):
                 log.log(0, "Skipping fetch for partition %s because there is an inflight request to node %s",
                         partition, node_id)
 
-        version = self._client.api_version(FetchRequest, max_version=4)
         requests = {}
         for node_id, partition_data in six.iteritems(fetchable):
-            if version < 3:
+            # As of version == 3 partitions will be returned in order as
+            # they are requested, so to avoid starvation with
+            # `fetch_max_bytes` option we need this shuffle
+            # NOTE: we do have partition_data in random order due to usage
+            #       of unordered structures like dicts, but that does not
+            #       guarantee equal distribution, and starting in Python3.6
+            #       dicts retain insert order.
+            partition_data = list(partition_data.items())
+            random.shuffle(partition_data)
+
+            if version <= 2:
                 requests[node_id] = FetchRequest[version](
                     -1,  # replica_id
                     self.config['fetch_max_wait_ms'],
                     self.config['fetch_min_bytes'],
-                    partition_data.items())
+                    partition_data)
+            elif version == 3:
+                requests[node_id] = FetchRequest[version](
+                    -1,  # replica_id
+                    self.config['fetch_max_wait_ms'],
+                    self.config['fetch_min_bytes'],
+                    self.config['fetch_max_bytes'],
+                    partition_data)
             else:
-                # As of version == 3 partitions will be returned in order as
-                # they are requested, so to avoid starvation with
-                # `fetch_max_bytes` option we need this shuffle
-                # NOTE: we do have partition_data in random order due to usage
-                #       of unordered structures like dicts, but that does not
-                #       guarantee equal distribution, and starting in Python3.6
-                #       dicts retain insert order.
-                partition_data = list(partition_data.items())
-                random.shuffle(partition_data)
-                if version == 3:
-                    requests[node_id] = FetchRequest[version](
-                        -1,  # replica_id
-                        self.config['fetch_max_wait_ms'],
-                        self.config['fetch_min_bytes'],
-                        self.config['fetch_max_bytes'],
-                        partition_data)
-                else:
-                    requests[node_id] = FetchRequest[version](
-                        -1,  # replica_id
-                        self.config['fetch_max_wait_ms'],
-                        self.config['fetch_min_bytes'],
-                        self.config['fetch_max_bytes'],
-                        self._isolation_level,
-                        partition_data)
+                # through v6
+                requests[node_id] = FetchRequest[version](
+                    -1,  # replica_id
+                    self.config['fetch_max_wait_ms'],
+                    self.config['fetch_min_bytes'],
+                    self.config['fetch_max_bytes'],
+                    self._isolation_level,
+                    partition_data)
         return requests
 
     def _handle_fetch_response(self, request, send_time, response):
@@ -821,7 +841,7 @@ class Fetcher(six.Iterator):
                     log.debug("Adding fetched record for partition %s with"
                               " offset %d to buffered record list", tp,
                               position)
-                    unpacked = list(self._unpack_message_set(tp, records))
+                    unpacked = list(self._unpack_records(tp, records))
                     parsed_records = self.PartitionRecords(fetch_offset, tp, unpacked)
                     if unpacked:
                         last_offset = unpacked[-1].offset
@@ -845,7 +865,9 @@ class Fetcher(six.Iterator):
                 self._sensors.record_topic_fetch_metrics(tp.topic, num_bytes, records_count)
 
             elif error_type in (Errors.NotLeaderForPartitionError,
-                                Errors.UnknownTopicOrPartitionError):
+                                Errors.UnknownTopicOrPartitionError,
+                                Errors.KafkaStorageError):
+                log.debug("Error fetching partition %s: %s", tp, error_type.__name__)
                 self._client.cluster.request_update()
             elif error_type is Errors.OffsetOutOfRangeError:
                 position = self._subscriptions.assignment[tp].position
@@ -862,8 +884,10 @@ class Fetcher(six.Iterator):
             elif error_type is Errors.TopicAuthorizationFailedError:
                 log.warning("Not authorized to read from topic %s.", tp.topic)
                 raise Errors.TopicAuthorizationFailedError(set(tp.topic))
-            elif error_type is Errors.UnknownError:
-                log.warning("Unknown error fetching data for topic-partition %s", tp)
+            elif error_type.is_retriable:
+                log.debug("Retriable error fetching partition %s: %s", tp, error_type())
+                if error_type.invalid_metadata:
+                    self._client.cluster.request_update()
             else:
                 raise error_type('Unexpected error while fetching data')
 
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index 06e10b8..38d7585 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -16,7 +16,7 @@ from kafka.coordinator.consumer import ConsumerCoordinator
 from kafka.coordinator.assignors.range import RangePartitionAssignor
 from kafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor
 from kafka.metrics import MetricConfig, Metrics
-from kafka.protocol.offset import OffsetResetStrategy
+from kafka.protocol.list_offsets import OffsetResetStrategy
 from kafka.structs import TopicPartition
 from kafka.version import __version__
 
diff --git a/kafka/consumer/subscription_state.py b/kafka/consumer/subscription_state.py
index 08842d1..a329ad3 100644
--- a/kafka/consumer/subscription_state.py
+++ b/kafka/consumer/subscription_state.py
@@ -7,7 +7,7 @@ import re
 from kafka.vendor import six
 
 from kafka.errors import IllegalStateError
-from kafka.protocol.offset import OffsetResetStrategy
+from kafka.protocol.list_offsets import OffsetResetStrategy
 from kafka.structs import OffsetAndMetadata
 
 log = logging.getLogger(__name__)
@@ -381,10 +381,10 @@ class TopicPartitionState(object):
         self.reset_strategy = None # the reset strategy if awaitingReset is set
         self._position = None # offset exposed to the user
         self.highwater = None
-        self.drop_pending_message_set = False
-        # The last message offset hint available from a message batch with
+        self.drop_pending_record_batch = False
+        # The last message offset hint available from a record batch with
         # magic=2 which includes deleted compacted messages
-        self.last_offset_from_message_batch = None
+        self.last_offset_from_record_batch = None
 
     def _set_position(self, offset):
         assert self.has_valid_position, 'Valid position required'
@@ -399,7 +399,7 @@ class TopicPartitionState(object):
         self.awaiting_reset = True
         self.reset_strategy = strategy
         self._position = None
-        self.last_offset_from_message_batch = None
+        self.last_offset_from_record_batch = None
         self.has_valid_position = False
 
     def seek(self, offset):
@@ -407,8 +407,8 @@ class TopicPartitionState(object):
         self.awaiting_reset = False
         self.reset_strategy = None
         self.has_valid_position = True
-        self.drop_pending_message_set = True
-        self.last_offset_from_message_batch = None
+        self.drop_pending_record_batch = True
+        self.last_offset_from_record_batch = None
 
     def pause(self):
         self.paused = True
diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py
index d18de07..8607e48 100644
--- a/kafka/coordinator/base.py
+++ b/kafka/coordinator/base.py
@@ -14,7 +14,8 @@ from kafka import errors as Errors
 from kafka.future import Future
 from kafka.metrics import AnonMeasurable
 from kafka.metrics.stats import Avg, Count, Max, Rate
-from kafka.protocol.commit import GroupCoordinatorRequest, OffsetCommitRequest
+from kafka.protocol.commit import OffsetCommitRequest
+from kafka.protocol.find_coordinator import FindCoordinatorRequest
 from kafka.protocol.group import (HeartbeatRequest, JoinGroupRequest,
                             LeaveGroupRequest, SyncGroupRequest)
 
@@ -452,7 +453,7 @@ class BaseCoordinator(object):
             (protocol, metadata if isinstance(metadata, bytes) else metadata.encode())
             for protocol, metadata in self.group_protocols()
         ]
-        version = self._client.api_version(JoinGroupRequest, max_version=2)
+        version = self._client.api_version(JoinGroupRequest, max_version=3)
         if version == 0:
             request = JoinGroupRequest[version](
                 self.group_id,
@@ -492,6 +493,11 @@ class BaseCoordinator(object):
         future.failure(error)
 
     def _handle_join_group_response(self, future, send_time, response):
+        if response.API_VERSION >= 2:
+            self.sensors.throttle_time.record(response.throttle_time_ms)
+            if response.throttle_time_ms > 0:
+                log.warning("JoinGroupRequest throttled by broker (%d ms)", response.throttle_time_ms)
+
         error_type = Errors.for_code(response.error_code)
         if error_type is Errors.NoError:
             log.debug("Received successful JoinGroup response for group %s: %s",
@@ -553,7 +559,7 @@ class BaseCoordinator(object):
 
     def _on_join_follower(self):
         # send follower's sync group with an empty assignment
-        version = self._client.api_version(SyncGroupRequest, max_version=1)
+        version = self._client.api_version(SyncGroupRequest, max_version=2)
         request = SyncGroupRequest[version](
             self.group_id,
             self._generation.generation_id,
@@ -581,7 +587,7 @@ class BaseCoordinator(object):
         except Exception as e:
             return Future().failure(e)
 
-        version = self._client.api_version(SyncGroupRequest, max_version=1)
+        version = self._client.api_version(SyncGroupRequest, max_version=2)
         request = SyncGroupRequest[version](
             self.group_id,
             self._generation.generation_id,
@@ -613,6 +619,11 @@ class BaseCoordinator(object):
         return future
 
     def _handle_sync_group_response(self, future, send_time, response):
+        if response.API_VERSION >= 1:
+            self.sensors.throttle_time.record(response.throttle_time_ms)
+            if response.throttle_time_ms > 0:
+                log.warning("SyncGroupRequest throttled by broker (%d ms)", response.throttle_time_ms)
+
         error_type = Errors.for_code(response.error_code)
         if error_type is Errors.NoError:
             self.sensors.sync_latency.record((time.time() - send_time) * 1000)
@@ -660,7 +671,11 @@ class BaseCoordinator(object):
 
         log.debug("Sending group coordinator request for group %s to broker %s",
                   self.group_id, node_id)
-        request = GroupCoordinatorRequest[0](self.group_id)
+        version = self._client.api_version(FindCoordinatorRequest, max_version=2)
+        if version == 0:
+            request = FindCoordinatorRequest[version](self.group_id)
+        else:
+            request = FindCoordinatorRequest[version](self.group_id, 0)
         future = Future()
         _f = self._client.send(node_id, request)
         _f.add_callback(self._handle_group_coordinator_response, future)
@@ -668,6 +683,9 @@ class BaseCoordinator(object):
         return future
 
     def _handle_group_coordinator_response(self, future, response):
+        if response.API_VERSION >= 1 and response.throttle_time_ms > 0:
+            log.warning("FindCoordinatorRequest throttled by broker (%d ms)", response.throttle_time_ms)
+
         log.debug("Received group coordinator response %s", response)
 
         error_type = Errors.for_code(response.error_code)
@@ -762,7 +780,7 @@ class BaseCoordinator(object):
                 # this is a minimal effort attempt to leave the group. we do not
                 # attempt any resending if the request fails or times out.
                 log.info('Leaving consumer group (%s).', self.group_id)
-                version = self._client.api_version(LeaveGroupRequest, max_version=1)
+                version = self._client.api_version(LeaveGroupRequest, max_version=2)
                 request = LeaveGroupRequest[version](self.group_id, self._generation.member_id)
                 future = self._client.send(self.coordinator_id, request)
                 future.add_callback(self._handle_leave_group_response)
@@ -772,6 +790,11 @@ class BaseCoordinator(object):
             self.reset_generation()
 
     def _handle_leave_group_response(self, response):
+        if response.API_VERSION >= 1:
+            self.sensors.throttle_time.record(response.throttle_time_ms)
+            if response.throttle_time_ms > 0:
+                log.warning("LeaveGroupRequest throttled by broker (%d ms)", response.throttle_time_ms)
+
         error_type = Errors.for_code(response.error_code)
         if error_type is Errors.NoError:
             log.debug("LeaveGroup request for group %s returned successfully",
@@ -790,7 +813,7 @@ class BaseCoordinator(object):
             e = Errors.NodeNotReadyError(self.coordinator_id)
             return Future().failure(e)
 
-        version = self._client.api_version(HeartbeatRequest, max_version=1)
+        version = self._client.api_version(HeartbeatRequest, max_version=2)
         request = HeartbeatRequest[version](self.group_id,
                                             self._generation.generation_id,
                                             self._generation.member_id)
@@ -803,6 +826,11 @@ class BaseCoordinator(object):
         return future
 
     def _handle_heartbeat_response(self, future, send_time, response):
+        if response.API_VERSION >= 1:
+            self.sensors.throttle_time.record(response.throttle_time_ms)
+            if response.throttle_time_ms > 0:
+                log.warning("HeartbeatRequest throttled by broker (%d ms)", response.throttle_time_ms)
+
         self.sensors.heartbeat_latency.record((time.time() - send_time) * 1000)
         error_type = Errors.for_code(response.error_code)
         if error_type is Errors.NoError:
@@ -891,6 +919,14 @@ class GroupCoordinatorMetrics(object):
             tags), AnonMeasurable(
                 lambda _, now: (now / 1000) - self.heartbeat.last_send))
 
+        self.throttle_time = metrics.sensor('throttle-time')
+        self.throttle_time.add(metrics.metric_name(
+            'throttle-time-avg', self.metric_group_name,
+            'The average throttle time in ms'), Avg())
+        self.throttle_time.add(metrics.metric_name(
+            'throttle-time-max', self.metric_group_name,
+            'The maximum throttle time in ms'), Max())
+
 
 class HeartbeatThread(threading.Thread):
     def __init__(self, coordinator):
diff --git a/kafka/producer/record_accumulator.py b/kafka/producer/record_accumulator.py
index a2aa0e8..f13c21b 100644
--- a/kafka/producer/record_accumulator.py
+++ b/kafka/producer/record_accumulator.py
@@ -156,7 +156,7 @@ class RecordAccumulator(object):
             will also impact the compression ratio (more batching means better
             compression). Default: None.
         linger_ms (int): An artificial delay time to add before declaring a
-            messageset (that isn't full) ready for sending. This allows
+            record batch (that isn't full) ready for sending. This allows
             time for more records to arrive. Setting a non-zero linger_ms
             will trade off some latency for potentially better throughput
             due to more batching (and hence fewer, larger requests).
diff --git a/kafka/protocol/commit.py b/kafka/protocol/commit.py
index 31fc237..f5828ba 100644
--- a/kafka/protocol/commit.py
+++ b/kafka/protocol/commit.py
@@ -207,49 +207,3 @@ OffsetFetchResponse = [
     OffsetFetchResponse_v0, OffsetFetchResponse_v1,
     OffsetFetchResponse_v2, OffsetFetchResponse_v3,
 ]
-
-
-class GroupCoordinatorResponse_v0(Response):
-    API_KEY = 10
-    API_VERSION = 0
-    SCHEMA = Schema(
-        ('error_code', Int16),
-        ('coordinator_id', Int32),
-        ('host', String('utf-8')),
-        ('port', Int32)
-    )
-
-
-class GroupCoordinatorResponse_v1(Response):
-    API_KEY = 10
-    API_VERSION = 1
-    SCHEMA = Schema(
-        ('error_code', Int16),
-        ('error_message', String('utf-8')),
-        ('coordinator_id', Int32),
-        ('host', String('utf-8')),
-        ('port', Int32)
-    )
-
-
-class GroupCoordinatorRequest_v0(Request):
-    API_KEY = 10
-    API_VERSION = 0
-    RESPONSE_TYPE = GroupCoordinatorResponse_v0
-    SCHEMA = Schema(
-        ('consumer_group', String('utf-8'))
-    )
-
-
-class GroupCoordinatorRequest_v1(Request):
-    API_KEY = 10
-    API_VERSION = 1
-    RESPONSE_TYPE = GroupCoordinatorResponse_v1
-    SCHEMA = Schema(
-        ('coordinator_key', String('utf-8')),
-        ('coordinator_type', Int8)
-    )
-
-
-GroupCoordinatorRequest = [GroupCoordinatorRequest_v0, GroupCoordinatorRequest_v1]
-GroupCoordinatorResponse = [GroupCoordinatorResponse_v0, GroupCoordinatorResponse_v1]
diff --git a/kafka/protocol/fetch.py b/kafka/protocol/fetch.py
index f367848..1b77e90 100644
--- a/kafka/protocol/fetch.py
+++ b/kafka/protocol/fetch.py
@@ -14,7 +14,7 @@ class FetchResponse_v0(Response):
                 ('partition', Int32),
                 ('error_code', Int16),
                 ('highwater_offset', Int64),
-                ('message_set', Bytes)))))
+                ('records', Bytes)))))
     )
 
 
@@ -29,7 +29,7 @@ class FetchResponse_v1(Response):
                 ('partition', Int32),
                 ('error_code', Int16),
                 ('highwater_offset', Int64),
-                ('message_set', Bytes)))))
+                ('records', Bytes)))))
     )
 
 
@@ -46,6 +46,7 @@ class FetchResponse_v3(Response):
 
 
 class FetchResponse_v4(Response):
+    # Adds message format v2
     API_KEY = 1
     API_VERSION = 4
     SCHEMA = Schema(
@@ -60,7 +61,7 @@ class FetchResponse_v4(Response):
                 ('aborted_transactions', Array(
                     ('producer_id', Int64),
                     ('first_offset', Int64))),
-                ('message_set', Bytes)))))
+                ('records', Bytes)))))
     )
 
 
@@ -80,7 +81,7 @@ class FetchResponse_v5(Response):
                 ('aborted_transactions', Array(
                     ('producer_id', Int64),
                     ('first_offset', Int64))),
-                ('message_set', Bytes)))))
+                ('records', Bytes)))))
     )
 
 
@@ -115,7 +116,7 @@ class FetchResponse_v7(Response):
                 ('aborted_transactions', Array(
                     ('producer_id', Int64),
                     ('first_offset', Int64))),
-                ('message_set', Bytes)))))
+                ('records', Bytes)))))
     )
 
 
@@ -156,7 +157,7 @@ class FetchResponse_v11(Response):
                     ('producer_id', Int64),
                     ('first_offset', Int64))),
                 ('preferred_read_replica', Int32),
-                ('message_set', Bytes)))))
+                ('records', Bytes)))))
     )
 
 
@@ -211,6 +212,7 @@ class FetchRequest_v3(Request):
 
 class FetchRequest_v4(Request):
     # Adds isolation_level field
+    # Adds message format v2
     API_KEY = 1
     API_VERSION = 4
     RESPONSE_TYPE = FetchResponse_v4
@@ -264,7 +266,7 @@ class FetchRequest_v6(Request):
 
 class FetchRequest_v7(Request):
     """
-    Add incremental fetch requests
+    Add incremental fetch requests (see KIP-227)
     """
     API_KEY = 1
     API_VERSION = 7
diff --git a/kafka/protocol/find_coordinator.py b/kafka/protocol/find_coordinator.py
new file mode 100644
index 0000000..a68a239
--- /dev/null
+++ b/kafka/protocol/find_coordinator.py
@@ -0,0 +1,64 @@
+from __future__ import absolute_import
+
+from kafka.protocol.api import Request, Response
+from kafka.protocol.types import Array, Int8, Int16, Int32, Int64, Schema, String
+
+
+class FindCoordinatorResponse_v0(Response):
+    API_KEY = 10
+    API_VERSION = 0
+    SCHEMA = Schema(
+        ('error_code', Int16),
+        ('coordinator_id', Int32),
+        ('host', String('utf-8')),
+        ('port', Int32)
+    )
+
+
+class FindCoordinatorResponse_v1(Response):
+    API_KEY = 10
+    API_VERSION = 1
+    SCHEMA = Schema(
+        ('throttle_time_ms', Int32),
+        ('error_code', Int16),
+        ('error_message', String('utf-8')),
+        ('coordinator_id', Int32),
+        ('host', String('utf-8')),
+        ('port', Int32)
+    )
+
+
+class FindCoordinatorResponse_v2(Response):
+    API_KEY = 10
+    API_VERSION = 2
+    SCHEMA = FindCoordinatorResponse_v1.SCHEMA
+
+
+class FindCoordinatorRequest_v0(Request):
+    API_KEY = 10
+    API_VERSION = 0
+    RESPONSE_TYPE = FindCoordinatorResponse_v0
+    SCHEMA = Schema(
+        ('consumer_group', String('utf-8'))
+    )
+
+
+class FindCoordinatorRequest_v1(Request):
+    API_KEY = 10
+    API_VERSION = 1
+    RESPONSE_TYPE = FindCoordinatorResponse_v1
+    SCHEMA = Schema(
+        ('coordinator_key', String('utf-8')),
+        ('coordinator_type', Int8) # 0: consumer, 1: transaction
+    )
+
+
+class FindCoordinatorRequest_v2(Request):
+    API_KEY = 10
+    API_VERSION = 2
+    RESPONSE_TYPE = FindCoordinatorResponse_v2
+    SCHEMA = FindCoordinatorRequest_v1.SCHEMA
+
+
+FindCoordinatorRequest = [FindCoordinatorRequest_v0, FindCoordinatorRequest_v1, FindCoordinatorRequest_v2]
+FindCoordinatorResponse = [FindCoordinatorResponse_v0, FindCoordinatorResponse_v1, FindCoordinatorResponse_v2]
diff --git a/kafka/protocol/group.py b/kafka/protocol/group.py
index bcb9655..ee06141 100644
--- a/kafka/protocol/group.py
+++ b/kafka/protocol/group.py
@@ -42,6 +42,12 @@ class JoinGroupResponse_v2(Response):
     )
 
 
+class JoinGroupResponse_v3(Response):
+    API_KEY = 11
+    API_VERSION = 3
+    SCHEMA = JoinGroupResponse_v2.SCHEMA
+
+
 class JoinGroupRequest_v0(Request):
     API_KEY = 11
     API_VERSION = 0
@@ -83,11 +89,19 @@ class JoinGroupRequest_v2(Request):
     UNKNOWN_MEMBER_ID = ''
 
 
+class JoinGroupRequest_v3(Request):
+    API_KEY = 11
+    API_VERSION = 3
+    RESPONSE_TYPE = JoinGroupResponse_v3
+    SCHEMA = JoinGroupRequest_v2.SCHEMA
+    UNKNOWN_MEMBER_ID = ''
+
+
 JoinGroupRequest = [
-    JoinGroupRequest_v0, JoinGroupRequest_v1, JoinGroupRequest_v2
+    JoinGroupRequest_v0, JoinGroupRequest_v1, JoinGroupRequest_v2, JoinGroupRequest_v3
 ]
 JoinGroupResponse = [
-    JoinGroupResponse_v0, JoinGroupResponse_v1, JoinGroupResponse_v2
+    JoinGroupResponse_v0, JoinGroupResponse_v1, JoinGroupResponse_v2, JoinGroupResponse_v3
 ]
 
 
@@ -118,6 +132,12 @@ class SyncGroupResponse_v1(Response):
     )
 
 
+class SyncGroupResponse_v2(Response):
+    API_KEY = 14
+    API_VERSION = 2
+    SCHEMA = SyncGroupResponse_v1.SCHEMA
+
+
 class SyncGroupRequest_v0(Request):
     API_KEY = 14
     API_VERSION = 0
@@ -139,8 +159,15 @@ class SyncGroupRequest_v1(Request):
     SCHEMA = SyncGroupRequest_v0.SCHEMA
 
 
-SyncGroupRequest = [SyncGroupRequest_v0, SyncGroupRequest_v1]
-SyncGroupResponse = [SyncGroupResponse_v0, SyncGroupResponse_v1]
+class SyncGroupRequest_v2(Request):
+    API_KEY = 14
+    API_VERSION = 2
+    RESPONSE_TYPE = SyncGroupResponse_v2
+    SCHEMA = SyncGroupRequest_v1.SCHEMA
+
+
+SyncGroupRequest = [SyncGroupRequest_v0, SyncGroupRequest_v1, SyncGroupRequest_v2]
+SyncGroupResponse = [SyncGroupResponse_v0, SyncGroupResponse_v1, SyncGroupResponse_v2]
 
 
 class MemberAssignment(Struct):
@@ -170,6 +197,12 @@ class HeartbeatResponse_v1(Response):
     )
 
 
+class HeartbeatResponse_v2(Response):
+    API_KEY = 12
+    API_VERSION = 2
+    SCHEMA = HeartbeatResponse_v1.SCHEMA
+
+
 class HeartbeatRequest_v0(Request):
     API_KEY = 12
     API_VERSION = 0
@@ -188,8 +221,15 @@ class HeartbeatRequest_v1(Request):
     SCHEMA = HeartbeatRequest_v0.SCHEMA
 
 
-HeartbeatRequest = [HeartbeatRequest_v0, HeartbeatRequest_v1]
-HeartbeatResponse = [HeartbeatResponse_v0, HeartbeatResponse_v1]
+class HeartbeatRequest_v2(Request):
+    API_KEY = 12
+    API_VERSION = 2
+    RESPONSE_TYPE = HeartbeatResponse_v2
+    SCHEMA = HeartbeatRequest_v1.SCHEMA
+
+
+HeartbeatRequest = [HeartbeatRequest_v0, HeartbeatRequest_v1, HeartbeatRequest_v2]
+HeartbeatResponse = [HeartbeatResponse_v0, HeartbeatResponse_v1, HeartbeatResponse_v2]
 
 
 class LeaveGroupResponse_v0(Response):
@@ -209,6 +249,12 @@ class LeaveGroupResponse_v1(Response):
     )
 
 
+class LeaveGroupResponse_v2(Response):
+    API_KEY = 13
+    API_VERSION = 2
+    SCHEMA = LeaveGroupResponse_v1.SCHEMA
+
+
 class LeaveGroupRequest_v0(Request):
     API_KEY = 13
     API_VERSION = 0
@@ -226,5 +272,12 @@ class LeaveGroupRequest_v1(Request):
     SCHEMA = LeaveGroupRequest_v0.SCHEMA
 
 
-LeaveGroupRequest = [LeaveGroupRequest_v0, LeaveGroupRequest_v1]
-LeaveGroupResponse = [LeaveGroupResponse_v0, LeaveGroupResponse_v1]
+class LeaveGroupRequest_v2(Request):
+    API_KEY = 13
+    API_VERSION = 2
+    RESPONSE_TYPE = LeaveGroupResponse_v2
+    SCHEMA = LeaveGroupRequest_v1.SCHEMA
+
+
+LeaveGroupRequest = [LeaveGroupRequest_v0, LeaveGroupRequest_v1, LeaveGroupRequest_v2]
+LeaveGroupResponse = [LeaveGroupResponse_v0, LeaveGroupResponse_v1, LeaveGroupResponse_v2]
diff --git a/kafka/protocol/offset.py b/kafka/protocol/list_offsets.py
similarity index 74%
rename from kafka/protocol/offset.py
rename to kafka/protocol/list_offsets.py
index 1ed382b..9c5ad5e 100644
--- a/kafka/protocol/offset.py
+++ b/kafka/protocol/list_offsets.py
@@ -12,7 +12,7 @@ class OffsetResetStrategy(object):
     NONE = 0
 
 
-class OffsetResponse_v0(Response):
+class ListOffsetsResponse_v0(Response):
     API_KEY = 2
     API_VERSION = 0
     SCHEMA = Schema(
@@ -24,7 +24,7 @@ class OffsetResponse_v0(Response):
                 ('offsets', Array(Int64))))))
     )
 
-class OffsetResponse_v1(Response):
+class ListOffsetsResponse_v1(Response):
     API_KEY = 2
     API_VERSION = 1
     SCHEMA = Schema(
@@ -38,7 +38,7 @@ class OffsetResponse_v1(Response):
     )
 
 
-class OffsetResponse_v2(Response):
+class ListOffsetsResponse_v2(Response):
     API_KEY = 2
     API_VERSION = 2
     SCHEMA = Schema(
@@ -53,16 +53,16 @@ class OffsetResponse_v2(Response):
     )
 
 
-class OffsetResponse_v3(Response):
+class ListOffsetsResponse_v3(Response):
     """
     on quota violation, brokers send out responses before throttling
     """
     API_KEY = 2
     API_VERSION = 3
-    SCHEMA = OffsetResponse_v2.SCHEMA
+    SCHEMA = ListOffsetsResponse_v2.SCHEMA
 
 
-class OffsetResponse_v4(Response):
+class ListOffsetsResponse_v4(Response):
     """
     Add leader_epoch to response
     """
@@ -81,19 +81,19 @@ class OffsetResponse_v4(Response):
     )
 
 
-class OffsetResponse_v5(Response):
+class ListOffsetsResponse_v5(Response):
     """
     adds a new error code, OFFSET_NOT_AVAILABLE
     """
     API_KEY = 2
     API_VERSION = 5
-    SCHEMA = OffsetResponse_v4.SCHEMA
+    SCHEMA = ListOffsetsResponse_v4.SCHEMA
 
 
-class OffsetRequest_v0(Request):
+class ListOffsetsRequest_v0(Request):
     API_KEY = 2
     API_VERSION = 0
-    RESPONSE_TYPE = OffsetResponse_v0
+    RESPONSE_TYPE = ListOffsetsResponse_v0
     SCHEMA = Schema(
         ('replica_id', Int32),
         ('topics', Array(
@@ -107,10 +107,10 @@ class OffsetRequest_v0(Request):
         'replica_id': -1
     }
 
-class OffsetRequest_v1(Request):
+class ListOffsetsRequest_v1(Request):
     API_KEY = 2
     API_VERSION = 1
-    RESPONSE_TYPE = OffsetResponse_v1
+    RESPONSE_TYPE = ListOffsetsResponse_v1
     SCHEMA = Schema(
         ('replica_id', Int32),
         ('topics', Array(
@@ -124,10 +124,10 @@ class OffsetRequest_v1(Request):
     }
 
 
-class OffsetRequest_v2(Request):
+class ListOffsetsRequest_v2(Request):
     API_KEY = 2
     API_VERSION = 2
-    RESPONSE_TYPE = OffsetResponse_v2
+    RESPONSE_TYPE = ListOffsetsResponse_v2
     SCHEMA = Schema(
         ('replica_id', Int32),
         ('isolation_level', Int8),  # <- added isolation_level
@@ -142,23 +142,23 @@ class OffsetRequest_v2(Request):
     }
 
 
-class OffsetRequest_v3(Request):
+class ListOffsetsRequest_v3(Request):
     API_KEY = 2
     API_VERSION = 3
-    RESPONSE_TYPE = OffsetResponse_v3
-    SCHEMA = OffsetRequest_v2.SCHEMA
+    RESPONSE_TYPE = ListOffsetsResponse_v3
+    SCHEMA = ListOffsetsRequest_v2.SCHEMA
     DEFAULTS = {
         'replica_id': -1
     }
 
 
-class OffsetRequest_v4(Request):
+class ListOffsetsRequest_v4(Request):
     """
     Add current_leader_epoch to request
     """
     API_KEY = 2
     API_VERSION = 4
-    RESPONSE_TYPE = OffsetResponse_v4
+    RESPONSE_TYPE = ListOffsetsResponse_v4
     SCHEMA = Schema(
         ('replica_id', Int32),
         ('isolation_level', Int8),  # <- added isolation_level
@@ -174,21 +174,21 @@ class OffsetRequest_v4(Request):
     }
 
 
-class OffsetRequest_v5(Request):
+class ListOffsetsRequest_v5(Request):
     API_KEY = 2
     API_VERSION = 5
-    RESPONSE_TYPE = OffsetResponse_v5
-    SCHEMA = OffsetRequest_v4.SCHEMA
+    RESPONSE_TYPE = ListOffsetsResponse_v5
+    SCHEMA = ListOffsetsRequest_v4.SCHEMA
     DEFAULTS = {
         'replica_id': -1
     }
 
 
-OffsetRequest = [
-    OffsetRequest_v0, OffsetRequest_v1, OffsetRequest_v2,
-    OffsetRequest_v3, OffsetRequest_v4, OffsetRequest_v5,
+ListOffsetsRequest = [
+    ListOffsetsRequest_v0, ListOffsetsRequest_v1, ListOffsetsRequest_v2,
+    ListOffsetsRequest_v3, ListOffsetsRequest_v4, ListOffsetsRequest_v5,
 ]
-OffsetResponse = [
-    OffsetResponse_v0, OffsetResponse_v1, OffsetResponse_v2,
-    OffsetResponse_v3, OffsetResponse_v4, OffsetResponse_v5,
+ListOffsetsResponse = [
+    ListOffsetsResponse_v0, ListOffsetsResponse_v1, ListOffsetsResponse_v2,
+    ListOffsetsResponse_v3, ListOffsetsResponse_v4, ListOffsetsResponse_v5,
 ]
diff --git a/kafka/protocol/parser.py b/kafka/protocol/parser.py
index a9e7672..e7799fc 100644
--- a/kafka/protocol/parser.py
+++ b/kafka/protocol/parser.py
@@ -4,7 +4,7 @@ import collections
 import logging
 
 import kafka.errors as Errors
-from kafka.protocol.commit import GroupCoordinatorResponse
+from kafka.protocol.find_coordinator import FindCoordinatorResponse
 from kafka.protocol.frame import KafkaBytes
 from kafka.protocol.types import Int32, TaggedFields
 from kafka.version import __version__
@@ -142,7 +142,7 @@ class KafkaProtocol(object):
         # 0.8.2 quirk
         if (recv_correlation_id == 0 and
             correlation_id != 0 and
-            request.RESPONSE_TYPE is GroupCoordinatorResponse[0] and
+            request.RESPONSE_TYPE is FindCoordinatorResponse[0] and
             (self._api_version == (0, 8, 2) or self._api_version is None)):
             log.warning('Kafka 0.8.2 quirk -- GroupCoordinatorResponse'
                         ' Correlation ID does not match request. This'
diff --git a/kafka/protocol/produce.py b/kafka/protocol/produce.py
index 9b3f6bf..3076a28 100644
--- a/kafka/protocol/produce.py
+++ b/kafka/protocol/produce.py
@@ -47,6 +47,7 @@ class ProduceResponse_v2(Response):
 
 
 class ProduceResponse_v3(Response):
+    # Adds support for message format v2
     API_KEY = 0
     API_VERSION = 3
     SCHEMA = ProduceResponse_v2.SCHEMA
@@ -141,7 +142,7 @@ class ProduceRequest_v0(ProduceRequest):
             ('topic', String('utf-8')),
             ('partitions', Array(
                 ('partition', Int32),
-                ('messages', Bytes)))))
+                ('records', Bytes)))))
     )
 
 
@@ -158,6 +159,7 @@ class ProduceRequest_v2(ProduceRequest):
 
 
 class ProduceRequest_v3(ProduceRequest):
+    # Adds support for message format v2
     API_VERSION = 3
     RESPONSE_TYPE = ProduceResponse_v3
     SCHEMA = Schema(
@@ -168,7 +170,7 @@ class ProduceRequest_v3(ProduceRequest):
             ('topic', String('utf-8')),
             ('partitions', Array(
                 ('partition', Int32),
-                ('messages', Bytes)))))
+                ('records', Bytes)))))
     )
 
 


Test patch:
diff --git a/test/test_fetcher.py b/test/test_fetcher.py
index bbc5b0c..e743692 100644
--- a/test/test_fetcher.py
+++ b/test/test_fetcher.py
@@ -18,7 +18,7 @@ from kafka.future import Future
 from kafka.metrics import Metrics
 from kafka.protocol.broker_api_versions import BROKER_API_VERSIONS
 from kafka.protocol.fetch import FetchRequest, FetchResponse
-from kafka.protocol.offset import OffsetResponse
+from kafka.protocol.list_offsets import ListOffsetsResponse
 from kafka.errors import (
     StaleMetadata, LeaderNotAvailableError, NotLeaderForPartitionError,
     UnknownTopicOrPartitionError, OffsetOutOfRangeError
@@ -149,9 +149,9 @@ def test__reset_offset(fetcher, mocker):
     assert fetcher._subscriptions.assignment[tp].position == 1001
 
 
-def test__send_offset_requests(fetcher, mocker):
-    tp = TopicPartition("topic_send_offset", 1)
-    mocked_send = mocker.patch.object(fetcher, "_send_offset_request")
+def test__send_list_offsets_requests(fetcher, mocker):
+    tp = TopicPartition("topic_send_list_offsets", 1)
+    mocked_send = mocker.patch.object(fetcher, "_send_list_offsets_request")
     send_futures = []
 
     def send_side_effect(*args, **kw):
@@ -168,19 +168,19 @@ def test__send_offset_requests(fetcher, mocker):
         [None, -1], itertools.cycle([0]))
 
     # Leader == None
-    fut = fetcher._send_offset_requests({tp: 0})
+    fut = fetcher._send_list_offsets_requests({tp: 0})
     assert fut.failed()
     assert isinstance(fut.exception, StaleMetadata)
     assert not mocked_send.called
 
     # Leader == -1
-    fut = fetcher._send_offset_requests({tp: 0})
+    fut = fetcher._send_list_offsets_requests({tp: 0})
     assert fut.failed()
     assert isinstance(fut.exception, LeaderNotAvailableError)
     assert not mocked_send.called
 
     # Leader == 0, send failed
-    fut = fetcher._send_offset_requests({tp: 0})
+    fut = fetcher._send_list_offsets_requests({tp: 0})
     assert not fut.is_done
     assert mocked_send.called
     # Check that we bound the futures correctly to chain failure
@@ -189,7 +189,7 @@ def test__send_offset_requests(fetcher, mocker):
     assert isinstance(fut.exception, NotLeaderForPartitionError)
 
     # Leader == 0, send success
-    fut = fetcher._send_offset_requests({tp: 0})
+    fut = fetcher._send_list_offsets_requests({tp: 0})
     assert not fut.is_done
     assert mocked_send.called
     # Check that we bound the futures correctly to chain success
@@ -198,12 +198,12 @@ def test__send_offset_requests(fetcher, mocker):
     assert fut.value == {tp: (10, 10000)}
 
 
-def test__send_offset_requests_multiple_nodes(fetcher, mocker):
-    tp1 = TopicPartition("topic_send_offset", 1)
-    tp2 = TopicPartition("topic_send_offset", 2)
-    tp3 = TopicPartition("topic_send_offset", 3)
-    tp4 = TopicPartition("topic_send_offset", 4)
-    mocked_send = mocker.patch.object(fetcher, "_send_offset_request")
+def test__send_list_offsets_requests_multiple_nodes(fetcher, mocker):
+    tp1 = TopicPartition("topic_send_list_offsets", 1)
+    tp2 = TopicPartition("topic_send_list_offsets", 2)
+    tp3 = TopicPartition("topic_send_list_offsets", 3)
+    tp4 = TopicPartition("topic_send_list_offsets", 4)
+    mocked_send = mocker.patch.object(fetcher, "_send_list_offsets_request")
     send_futures = []
 
     def send_side_effect(node_id, timestamps):
@@ -218,7 +218,7 @@ def test__send_offset_requests_multiple_nodes(fetcher, mocker):
 
     # -- All node succeeded case
     tss = OrderedDict([(tp1, 0), (tp2, 0), (tp3, 0), (tp4, 0)])
-    fut = fetcher._send_offset_requests(tss)
+    fut = fetcher._send_list_offsets_requests(tss)
     assert not fut.is_done
     assert mocked_send.call_count == 2
 
@@ -244,7 +244,7 @@ def test__send_offset_requests_multiple_nodes(fetcher, mocker):
 
     # -- First succeeded second not
     del send_futures[:]
-    fut = fetcher._send_offset_requests(tss)
+    fut = fetcher._send_list_offsets_requests(tss)
     assert len(send_futures) == 2
     send_futures[0][2].success({tp1: (11, 1001)})
     send_futures[1][2].failure(UnknownTopicOrPartitionError(tp1))
@@ -253,7 +253,7 @@ def test__send_offset_requests_multiple_nodes(fetcher, mocker):
 
     # -- First fails second succeeded
     del send_futures[:]
-    fut = fetcher._send_offset_requests(tss)
+    fut = fetcher._send_list_offsets_requests(tss)
     assert len(send_futures) == 2
     send_futures[0][2].failure(UnknownTopicOrPartitionError(tp1))
     send_futures[1][2].success({tp1: (11, 1001)})
@@ -261,49 +261,71 @@ def test__send_offset_requests_multiple_nodes(fetcher, mocker):
     assert isinstance(fut.exception, UnknownTopicOrPartitionError)
 
 
-def test__handle_offset_response(fetcher, mocker):
+def test__handle_list_offsets_response_v1(fetcher, mocker):
     # Broker returns UnsupportedForMessageFormatError, will omit partition
     fut = Future()
-    res = OffsetResponse[1]([
+    res = ListOffsetsResponse[1]([
         ("topic", [(0, 43, -1, -1)]),
         ("topic", [(1, 0, 1000, 9999)])
     ])
-    fetcher._handle_offset_response(fut, res)
+    fetcher._handle_list_offsets_response(fut, res)
     assert fut.succeeded()
     assert fut.value == {TopicPartition("topic", 1): (9999, 1000)}
 
     # Broker returns NotLeaderForPartitionError
     fut = Future()
-    res = OffsetResponse[1]([
+    res = ListOffsetsResponse[1]([
         ("topic", [(0, 6, -1, -1)]),
     ])
-    fetcher._handle_offset_response(fut, res)
+    fetcher._handle_list_offsets_response(fut, res)
     assert fut.failed()
     assert isinstance(fut.exception, NotLeaderForPartitionError)
 
     # Broker returns UnknownTopicOrPartitionError
     fut = Future()
-    res = OffsetResponse[1]([
+    res = ListOffsetsResponse[1]([
         ("topic", [(0, 3, -1, -1)]),
     ])
-    fetcher._handle_offset_response(fut, res)
+    fetcher._handle_list_offsets_response(fut, res)
     assert fut.failed()
     assert isinstance(fut.exception, UnknownTopicOrPartitionError)
 
     # Broker returns many errors and 1 result
     # Will fail on 1st error and return
     fut = Future()
-    res = OffsetResponse[1]([
+    res = ListOffsetsResponse[1]([
         ("topic", [(0, 43, -1, -1)]),
         ("topic", [(1, 6, -1, -1)]),
         ("topic", [(2, 3, -1, -1)]),
         ("topic", [(3, 0, 1000, 9999)])
     ])
-    fetcher._handle_offset_response(fut, res)
+    fetcher._handle_list_offsets_response(fut, res)
     assert fut.failed()
     assert isinstance(fut.exception, NotLeaderForPartitionError)
 
 
+def test__handle_list_offsets_response_v2_v3(fetcher, mocker):
+    # including a throttle_time shouldnt cause issues
+    fut = Future()
+    res = ListOffsetsResponse[2](
+        123, # throttle_time_ms
+        [("topic", [(0, 0, 1000, 9999)])
+    ])
+    fetcher._handle_list_offsets_response(fut, res)
+    assert fut.succeeded()
+    assert fut.value == {TopicPartition("topic", 0): (9999, 1000)}
+
+    # v3 response is the same format
+    fut = Future()
+    res = ListOffsetsResponse[3](
+        123, # throttle_time_ms
+        [("topic", [(0, 0, 1000, 9999)])
+    ])
+    fetcher._handle_list_offsets_response(fut, res)
+    assert fut.succeeded()
+    assert fut.value == {TopicPartition("topic", 0): (9999, 1000)}
+
+
 def test_fetched_records(fetcher, topic, mocker):
     fetcher.config['check_crcs'] = False
     tp = TopicPartition(topic, 0)
@@ -399,7 +421,7 @@ def test__handle_fetch_error(fetcher, caplog, exception, log_level):
     assert caplog.records[0].levelname == logging.getLevelName(log_level)
 
 
-def test__unpack_message_set(fetcher):
+def test__unpack_records(fetcher):
     fetcher.config['check_crcs'] = False
     tp = TopicPartition('foo', 0)
     messages = [
@@ -408,7 +430,7 @@ def test__unpack_message_set(fetcher):
         (None, b"c", None),
     ]
     memory_records = MemoryRecords(_build_record_batch(messages))
-    records = list(fetcher._unpack_message_set(tp, memory_records))
+    records = list(fetcher._unpack_records(tp, memory_records))
     assert len(records) == 3
     assert all(map(lambda x: isinstance(x, ConsumerRecord), records))
     assert records[0].value == b'a'
diff --git a/test/test_protocol.py b/test/test_protocol.py
index 6a77e19..6f94c74 100644
--- a/test/test_protocol.py
+++ b/test/test_protocol.py
@@ -5,8 +5,8 @@ import struct
 import pytest
 
 from kafka.protocol.api import RequestHeader
-from kafka.protocol.commit import GroupCoordinatorRequest
 from kafka.protocol.fetch import FetchRequest, FetchResponse
+from kafka.protocol.find_coordinator import FindCoordinatorRequest
 from kafka.protocol.message import Message, MessageSet, PartialMessage
 from kafka.protocol.metadata import MetadataRequest
 from kafka.protocol.types import Int16, Int32, Int64, String, UnsignedVarInt32, CompactString, CompactArray, CompactBytes
@@ -168,7 +168,7 @@ def test_encode_message_header():
         b'client3',                        # ClientId
     ])
 
-    req = GroupCoordinatorRequest[0]('foo')
+    req = FindCoordinatorRequest[0]('foo')
     header = RequestHeader(req, correlation_id=4, client_id='client3')
     assert header.encode() == expect