Repository: dpkp/kafka-python
Pull request #1980 (2019-12-29 20:15:38)

Do not log topic-specific errors in full metadata fetch
See #1088

<!-- Reviewable:start -->
---
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/dpkp/kafka-python/1980)
<!-- Reviewable:end -->



Patch:
diff --git a/kafka/admin/acl_resource.py b/kafka/admin/acl_resource.py
index 7a012d2..fd997a1 100644
--- a/kafka/admin/acl_resource.py
+++ b/kafka/admin/acl_resource.py
@@ -112,6 +112,24 @@ class ACLFilter(object):
             resource=self.resource_pattern
         )
 
+    def __eq__(self, other):
+        return all((
+            self.principal == other.principal,
+            self.host == other.host,
+            self.operation == other.operation,
+            self.permission_type == other.permission_type,
+            self.resource_pattern == other.resource_pattern
+        ))
+
+    def __hash__(self):
+        return hash((
+            self.principal,
+            self.host,
+            self.operation,
+            self.permission_type,
+            self.resource_pattern,
+        ))
+
 
 class ACL(ACLFilter):
     """Represents a concrete ACL for a specific ResourcePattern
@@ -181,6 +199,20 @@ class ResourcePatternFilter(object):
             self.pattern_type.name
         )
 
+    def __eq__(self, other):
+        return all((
+            self.resource_type == other.resource_type,
+            self.resource_name == other.resource_name,
+            self.pattern_type == other.pattern_type,
+        ))
+
+    def __hash__(self):
+        return hash((
+            self.resource_type,
+            self.resource_name,
+            self.pattern_type
+        ))
+
 
 class ResourcePattern(ResourcePatternFilter):
     """A resource pattern to apply the ACL to
@@ -209,4 +241,4 @@ class ResourcePattern(ResourcePatternFilter):
         if self.pattern_type in [ACLResourcePatternType.ANY, ACLResourcePatternType.MATCH]:
             raise IllegalArgumentError(
                 "pattern_type cannot be {} on a concrete ResourcePattern".format(self.pattern_type.name)
-            )
\ No newline at end of file
+            )
diff --git a/kafka/admin/client.py b/kafka/admin/client.py
index 8afe95b..e9be1d8 100644
--- a/kafka/admin/client.py
+++ b/kafka/admin/client.py
@@ -204,6 +204,7 @@ class KafkaAdminClient(object):
         self._client = KafkaClient(metrics=self._metrics,
                                    metric_group_prefix='admin',
                                    **self.config)
+        self._client.check_version()
 
         # Get auto-discovered version from client if necessary
         if self.config['api_version'] is None:
@@ -435,7 +436,7 @@ class KafkaAdminClient(object):
                 create_topic_requests=[self._convert_new_topic_request(new_topic) for new_topic in new_topics],
                 timeout=timeout_ms
             )
-        elif version <= 2:
+        elif version <= 3:
             request = CreateTopicsRequest[version](
                 create_topic_requests=[self._convert_new_topic_request(new_topic) for new_topic in new_topics],
                 timeout=timeout_ms,
@@ -459,7 +460,7 @@ class KafkaAdminClient(object):
         """
         version = self._matching_api_version(DeleteTopicsRequest)
         timeout_ms = self._validate_timeout(timeout_ms)
-        if version <= 1:
+        if version <= 3:
             request = DeleteTopicsRequest[version](
                 topics=topics,
                 timeout=timeout_ms
@@ -803,7 +804,7 @@ class KafkaAdminClient(object):
                     DescribeConfigsRequest[version](resources=topic_resources)
                 ))
 
-        elif version == 1:
+        elif version <= 2:
             if len(broker_resources) > 0:
                 for broker_resource in broker_resources:
                     try:
@@ -853,7 +854,7 @@ class KafkaAdminClient(object):
         :return: Appropriate version of AlterConfigsResponse class.
         """
         version = self._matching_api_version(AlterConfigsRequest)
-        if version == 0:
+        if version <= 1:
             request = AlterConfigsRequest[version](
                 resources=[self._convert_alter_config_resource_request(config_resource) for config_resource in config_resources]
             )
@@ -901,7 +902,7 @@ class KafkaAdminClient(object):
         """
         version = self._matching_api_version(CreatePartitionsRequest)
         timeout_ms = self._validate_timeout(timeout_ms)
-        if version == 0:
+        if version <= 1:
             request = CreatePartitionsRequest[version](
                 topic_partitions=[self._convert_create_partitions_request(topic_name, new_partitions) for topic_name, new_partitions in topic_partitions.items()],
                 timeout=timeout_ms,
@@ -928,7 +929,7 @@ class KafkaAdminClient(object):
     # describe delegation_token protocol not yet implemented
     # Note: send the request to the least_loaded_node()
 
-    def _describe_consumer_groups_send_request(self, group_id, group_coordinator_id):
+    def _describe_consumer_groups_send_request(self, group_id, group_coordinator_id, include_authorized_operations=False):
         """Send a DescribeGroupsRequest to the group's coordinator.
 
         :param group_id: The group name as a string
@@ -937,13 +938,24 @@ class KafkaAdminClient(object):
         :return: A message future.
         """
         version = self._matching_api_version(DescribeGroupsRequest)
-        if version <= 1:
+        if version <= 2:
+            if include_authorized_operations:
+                raise IncompatibleBrokerVersion(
+                    "include_authorized_operations requests "
+                    "DescribeGroupsRequest >= v3, which is not "
+                    "supported by Kafka {}".format(version)
+                )
             # Note: KAFKA-6788 A potential optimization is to group the
             # request per coordinator and send one request with a list of
             # all consumer groups. Java still hasn't implemented this
             # because the error checking is hard to get right when some
             # groups error and others don't.
             request = DescribeGroupsRequest[version](groups=(group_id,))
+        elif version <= 3:
+            request = DescribeGroupsRequest[version](
+                groups=(group_id,),
+                include_authorized_operations=include_authorized_operations
+            )
         else:
             raise NotImplementedError(
                 "Support for DescribeGroupsRequest_v{} has not yet been added to KafkaAdminClient."
@@ -952,7 +964,7 @@ class KafkaAdminClient(object):
 
     def _describe_consumer_groups_process_response(self, response):
         """Process a DescribeGroupsResponse into a group description."""
-        if response.API_VERSION <= 1:
+        if response.API_VERSION <= 3:
             assert len(response.groups) == 1
             # TODO need to implement converting the response tuple into
             # a more accessible interface like a namedtuple and then stop
@@ -976,7 +988,7 @@ class KafkaAdminClient(object):
                 .format(response.API_VERSION))
         return group_description
 
-    def describe_consumer_groups(self, group_ids, group_coordinator_id=None):
+    def describe_consumer_groups(self, group_ids, group_coordinator_id=None, include_authorized_operations=False):
         """Describe a set of consumer groups.
 
         Any errors are immediately raised.
@@ -989,6 +1001,9 @@ class KafkaAdminClient(object):
             useful for avoiding extra network round trips if you already know
             the group coordinator. This is only useful when all the group_ids
             have the same coordinator, otherwise it will error. Default: None.
+        :param include_authorized_operations: Whether or not to include
+            information about the operations a group is allowed to perform.
+            Only supported on API version >= v3. Default: False.
         :return: A list of group descriptions. For now the group descriptions
             are the raw results from the DescribeGroupsResponse. Long-term, we
             plan to change this to return namedtuples as well as decoding the
@@ -1001,7 +1016,10 @@ class KafkaAdminClient(object):
                 this_groups_coordinator_id = group_coordinator_id
             else:
                 this_groups_coordinator_id = self._find_coordinator_id(group_id)
-            f = self._describe_consumer_groups_send_request(group_id, this_groups_coordinator_id)
+            f = self._describe_consumer_groups_send_request(
+                group_id,
+                this_groups_coordinator_id,
+                include_authorized_operations)
             futures.append(f)
 
         self._wait_for_futures(futures)
diff --git a/kafka/cluster.py b/kafka/cluster.py
index 19137de..438baf2 100644
--- a/kafka/cluster.py
+++ b/kafka/cluster.py
@@ -285,6 +285,10 @@ class ClusterMetadata(object):
                         _new_broker_partitions[leader].add(
                             TopicPartition(topic, partition))
 
+            # Specific topic errors can be ignored if this is a full metadata fetch
+            elif self.need_all_topic_metadata:
+                continue
+
             elif error_type is Errors.LeaderNotAvailableError:
                 log.warning("Topic %s is not available during auto-create"
                             " initialization", topic)
diff --git a/kafka/conn.py b/kafka/conn.py
index dfb8d78..c383123 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -1203,6 +1203,10 @@ class BrokerConnection(object):
             stashed[key] = self.config[key]
             self.config[key] = override_config[key]
 
+        def reset_override_configs():
+            for key in stashed:
+                self.config[key] = stashed[key]
+
         # kafka kills the connection when it doesn't recognize an API request
         # so we can send a test request and then follow immediately with a
         # vanilla MetadataRequest. If the server did not recognize the first
@@ -1222,6 +1226,7 @@ class BrokerConnection(object):
 
         for version, request in test_cases:
             if not self.connect_blocking(timeout_at - time.time()):
+                reset_override_configs()
                 raise Errors.NodeNotReadyError()
             f = self.send(request)
             # HACK: sleeping to wait for socket to send bytes
@@ -1278,10 +1283,10 @@ class BrokerConnection(object):
             log.info("Broker is not v%s -- it did not recognize %s",
                      version, request.__class__.__name__)
         else:
+            reset_override_configs()
             raise Errors.UnrecognizedBrokerVersion()
 
-        for key in stashed:
-            self.config[key] = stashed[key]
+        reset_override_configs()
         return version
 
     def __str__(self):
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 8474b7c..458e9fd 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
@@ -638,6 +645,7 @@ class KafkaConsumer(six.Iterator):
             max_records = self.config['max_poll_records']
         assert isinstance(max_records, int), 'max_records must be an integer'
         assert max_records > 0, 'max_records must be positive'
+        assert not self._closed, 'KafkaConsumer is closed'
 
         # Poll for new data until the timeout expires
         start = time.time()
@@ -1173,6 +1181,8 @@ class KafkaConsumer(six.Iterator):
         return self
 
     def __next__(self):
+        if self._closed:
+            raise StopIteration('KafkaConsumer closed')
         # Now that the heartbeat thread runs in the background
         # there should be no reason to maintain a separate iterator
         # but we'll keep it available for a few releases just in case
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
diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py
index d91f3ea..b0e236a 100644
--- a/kafka/coordinator/base.py
+++ b/kafka/coordinator/base.py
@@ -489,13 +489,16 @@ class BaseCoordinator(object):
         return future
 
     def _failed_request(self, node_id, request, future, error):
-        log.error('Error sending %s to node %s [%s]',
-                  request.__class__.__name__, node_id, error)
         # Marking coordinator dead
         # unless the error is caused by internal client pipelining
         if not isinstance(error, (Errors.NodeNotReadyError,
                                   Errors.TooManyInFlightRequests)):
+            log.error('Error sending %s to node %s [%s]',
+                      request.__class__.__name__, node_id, error)
             self.coordinator_dead(error)
+        else:
+            log.debug('Error sending %s to node %s [%s]',
+                      request.__class__.__name__, node_id, error)
         future.failure(error)
 
     def _handle_join_group_response(self, future, send_time, response):
diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index 30337c3..fda80aa 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -387,7 +387,7 @@ class ConsumerCoordinator(BaseCoordinator):
             for partition, offset in six.iteritems(offsets):
                 # verify assignment is still active
                 if self._subscription.is_assigned(partition):
-                    self._subscription.assignment[partition].committed = offset.offset
+                    self._subscription.assignment[partition].committed = offset
             self._subscription.needs_fetch_committed_offsets = False
 
     def fetch_committed_offsets(self, partitions):
@@ -641,7 +641,7 @@ class ConsumerCoordinator(BaseCoordinator):
                     log.debug("Group %s committed offset %s for partition %s",
                               self.group_id, offset, tp)
                     if self._subscription.is_assigned(tp):
-                        self._subscription.assignment[tp].committed = offset.offset
+                        self._subscription.assignment[tp].committed = offset
                 elif error_type is Errors.GroupAuthorizationFailedError:
                     log.error("Not authorized to commit offsets for group %s",
                               self.group_id)
@@ -704,7 +704,7 @@ class ConsumerCoordinator(BaseCoordinator):
             partitions (list of TopicPartition): the partitions to fetch
 
         Returns:
-            Future: resolves to dict of offsets: {TopicPartition: int}
+            Future: resolves to dict of offsets: {TopicPartition: OffsetAndMetadata}
         """
         assert self.config['api_version'] >= (0, 8, 1), 'Unsupported Broker API'
         assert all(map(lambda k: isinstance(k, TopicPartition), partitions))
diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
index b90ca88..9509ab9 100644
--- a/kafka/producer/kafka.py
+++ b/kafka/producer/kafka.py
@@ -445,7 +445,7 @@ class KafkaProducer(object):
         self._cleanup = None
 
     def __del__(self):
-        self.close(timeout=0)
+        self.close()
 
     def close(self, timeout=None):
         """Close this producer.
@@ -484,14 +484,10 @@ class KafkaProducer(object):
                     self._sender.join(timeout)
 
         if self._sender is not None and self._sender.is_alive():
-
             log.info("Proceeding to force close the producer since pending"
                      " requests could not be completed within timeout %s.",
                      timeout)
             self._sender.force_close()
-            # Only join the sender thread when not calling from callback.
-            if not invoked_from_callback:
-                self._sender.join()
 
         self._metrics.close()
         try:
diff --git a/kafka/protocol/admin.py b/kafka/protocol/admin.py
index e6efad7..b2694dc 100644
--- a/kafka/protocol/admin.py
+++ b/kafka/protocol/admin.py
@@ -1,7 +1,7 @@
 from __future__ import absolute_import
 
 from kafka.protocol.api import Request, Response
-from kafka.protocol.types import Array, Boolean, Bytes, Int8, Int16, Int32, Schema, String
+from kafka.protocol.types import Array, Boolean, Bytes, Int8, Int16, Int32, Int64, Schema, String
 
 
 class ApiVersionResponse_v0(Response):
@@ -29,6 +29,12 @@ class ApiVersionResponse_v1(Response):
     )
 
 
+class ApiVersionResponse_v2(Response):
+    API_KEY = 18
+    API_VERSION = 2
+    SCHEMA = ApiVersionResponse_v1.SCHEMA
+
+
 class ApiVersionRequest_v0(Request):
     API_KEY = 18
     API_VERSION = 0
@@ -43,8 +49,19 @@ class ApiVersionRequest_v1(Request):
     SCHEMA = ApiVersionRequest_v0.SCHEMA
 
 
-ApiVersionRequest = [ApiVersionRequest_v0, ApiVersionRequest_v1]
-ApiVersionResponse = [ApiVersionResponse_v0, ApiVersionResponse_v1]
+class ApiVersionRequest_v2(Request):
+    API_KEY = 18
+    API_VERSION = 2
+    RESPONSE_TYPE = ApiVersionResponse_v1
+    SCHEMA = ApiVersionRequest_v0.SCHEMA
+
+
+ApiVersionRequest = [
+    ApiVersionRequest_v0, ApiVersionRequest_v1, ApiVersionRequest_v2,
+]
+ApiVersionResponse = [
+    ApiVersionResponse_v0, ApiVersionResponse_v1, ApiVersionResponse_v2,
+]
 
 
 class CreateTopicsResponse_v0(Response):
@@ -79,6 +96,11 @@ class CreateTopicsResponse_v2(Response):
             ('error_message', String('utf-8'))))
     )
 
+class CreateTopicsResponse_v3(Response):
+    API_KEY = 19
+    API_VERSION = 3
+    SCHEMA = CreateTopicsResponse_v2.SCHEMA
+
 
 class CreateTopicsRequest_v0(Request):
     API_KEY = 19
@@ -126,11 +148,20 @@ class CreateTopicsRequest_v2(Request):
     SCHEMA = CreateTopicsRequest_v1.SCHEMA
 
 
+class CreateTopicsRequest_v3(Request):
+    API_KEY = 19
+    API_VERSION = 3
+    RESPONSE_TYPE = CreateTopicsResponse_v3
+    SCHEMA = CreateTopicsRequest_v1.SCHEMA
+
+
 CreateTopicsRequest = [
-    CreateTopicsRequest_v0, CreateTopicsRequest_v1, CreateTopicsRequest_v2
+    CreateTopicsRequest_v0, CreateTopicsRequest_v1,
+    CreateTopicsRequest_v2, CreateTopicsRequest_v3,
 ]
 CreateTopicsResponse = [
-    CreateTopicsResponse_v0, CreateTopicsResponse_v1, CreateTopicsResponse_v2
+    CreateTopicsResponse_v0, CreateTopicsResponse_v1,
+    CreateTopicsResponse_v2, CreateTopicsResponse_v3,
 ]
 
 
@@ -155,6 +186,18 @@ class DeleteTopicsResponse_v1(Response):
     )
 
 
+class DeleteTopicsResponse_v2(Response):
+    API_KEY = 20
+    API_VERSION = 2
+    SCHEMA = DeleteTopicsResponse_v1.SCHEMA
+
+
+class DeleteTopicsResponse_v3(Response):
+    API_KEY = 20
+    API_VERSION = 3
+    SCHEMA = DeleteTopicsResponse_v1.SCHEMA
+
+
 class DeleteTopicsRequest_v0(Request):
     API_KEY = 20
     API_VERSION = 0
@@ -172,8 +215,28 @@ class DeleteTopicsRequest_v1(Request):
     SCHEMA = DeleteTopicsRequest_v0.SCHEMA
 
 
-DeleteTopicsRequest = [DeleteTopicsRequest_v0, DeleteTopicsRequest_v1]
-DeleteTopicsResponse = [DeleteTopicsResponse_v0, DeleteTopicsResponse_v1]
+class DeleteTopicsRequest_v2(Request):
+    API_KEY = 20
+    API_VERSION = 2
+    RESPONSE_TYPE = DeleteTopicsResponse_v2
+    SCHEMA = DeleteTopicsRequest_v0.SCHEMA
+
+
+class DeleteTopicsRequest_v3(Request):
+    API_KEY = 20
+    API_VERSION = 3
+    RESPONSE_TYPE = DeleteTopicsResponse_v3
+    SCHEMA = DeleteTopicsRequest_v0.SCHEMA
+
+
+DeleteTopicsRequest = [
+    DeleteTopicsRequest_v0, DeleteTopicsRequest_v1,
+    DeleteTopicsRequest_v2, DeleteTopicsRequest_v3,
+]
+DeleteTopicsResponse = [
+    DeleteTopicsResponse_v0, DeleteTopicsResponse_v1,
+    DeleteTopicsResponse_v2, DeleteTopicsResponse_v3,
+]
 
 
 class ListGroupsResponse_v0(Response):
@@ -198,6 +261,11 @@ class ListGroupsResponse_v1(Response):
             ('protocol_type', String('utf-8'))))
     )
 
+class ListGroupsResponse_v2(Response):
+    API_KEY = 16
+    API_VERSION = 2
+    SCHEMA = ListGroupsResponse_v1.SCHEMA
+
 
 class ListGroupsRequest_v0(Request):
     API_KEY = 16
@@ -212,9 +280,21 @@ class ListGroupsRequest_v1(Request):
     RESPONSE_TYPE = ListGroupsResponse_v1
     SCHEMA = ListGroupsRequest_v0.SCHEMA
 
+class ListGroupsRequest_v2(Request):
+    API_KEY = 16
+    API_VERSION = 1
+    RESPONSE_TYPE = ListGroupsResponse_v2
+    SCHEMA = ListGroupsRequest_v0.SCHEMA
 
-ListGroupsRequest = [ListGroupsRequest_v0, ListGroupsRequest_v1]
-ListGroupsResponse = [ListGroupsResponse_v0, ListGroupsResponse_v1]
+
+ListGroupsRequest = [
+    ListGroupsRequest_v0, ListGroupsRequest_v1,
+    ListGroupsRequest_v2,
+]
+ListGroupsResponse = [
+    ListGroupsResponse_v0, ListGroupsResponse_v1,
+    ListGroupsResponse_v2,
+]
 
 
 class DescribeGroupsResponse_v0(Response):
@@ -256,6 +336,33 @@ class DescribeGroupsResponse_v1(Response):
     )
 
 
+class DescribeGroupsResponse_v2(Response):
+    API_KEY = 15
+    API_VERSION = 2
+    SCHEMA = DescribeGroupsResponse_v1.SCHEMA
+
+
+class DescribeGroupsResponse_v3(Response):
+    API_KEY = 15
+    API_VERSION = 3
+    SCHEMA = Schema(
+        ('throttle_time_ms', Int32),
+        ('groups', Array(
+            ('error_code', Int16),
+            ('group', String('utf-8')),
+            ('state', String('utf-8')),
+            ('protocol_type', String('utf-8')),
+            ('protocol', String('utf-8')),
+            ('members', Array(
+                ('member_id', String('utf-8')),
+                ('client_id', String('utf-8')),
+                ('client_host', String('utf-8')),
+                ('member_metadata', Bytes),
+                ('member_assignment', Bytes)))),
+            ('authorized_operations', Int32))
+    )
+
+
 class DescribeGroupsRequest_v0(Request):
     API_KEY = 15
     API_VERSION = 0
@@ -272,8 +379,31 @@ class DescribeGroupsRequest_v1(Request):
     SCHEMA = DescribeGroupsRequest_v0.SCHEMA
 
 
-DescribeGroupsRequest = [DescribeGroupsRequest_v0, DescribeGroupsRequest_v1]
-DescribeGroupsResponse = [DescribeGroupsResponse_v0, DescribeGroupsResponse_v1]
+class DescribeGroupsRequest_v2(Request):
+    API_KEY = 15
+    API_VERSION = 2
+    RESPONSE_TYPE = DescribeGroupsResponse_v2
+    SCHEMA = DescribeGroupsRequest_v0.SCHEMA
+
+
+class DescribeGroupsRequest_v3(Request):
+    API_KEY = 15
+    API_VERSION = 3
+    RESPONSE_TYPE = DescribeGroupsResponse_v2
+    SCHEMA = Schema(
+        ('groups', Array(String('utf-8'))),
+        ('include_authorized_operations', Boolean)
+    )
+
+
+DescribeGroupsRequest = [
+    DescribeGroupsRequest_v0, DescribeGroupsRequest_v1,
+    DescribeGroupsRequest_v2, DescribeGroupsRequest_v3,
+]
+DescribeGroupsResponse = [
+    DescribeGroupsResponse_v0, DescribeGroupsResponse_v1,
+    DescribeGroupsResponse_v2, DescribeGroupsResponse_v3,
+]
 
 
 class SaslHandShakeResponse_v0(Response):
@@ -507,6 +637,13 @@ class AlterConfigsResponse_v0(Response):
             ('resource_name', String('utf-8'))))
     )
 
+
+class AlterConfigsResponse_v1(Response):
+    API_KEY = 33
+    API_VERSION = 1
+    SCHEMA = AlterConfigsResponse_v0.SCHEMA
+
+
 class AlterConfigsRequest_v0(Request):
     API_KEY = 33
     API_VERSION = 0
@@ -521,8 +658,14 @@ class AlterConfigsRequest_v0(Request):
         ('validate_only', Boolean)
     )
 
-AlterConfigsRequest = [AlterConfigsRequest_v0]
-AlterConfigsResponse = [AlterConfigsResponse_v0]
+class AlterConfigsRequest_v1(Request):
+    API_KEY = 33
+    API_VERSION = 1
+    RESPONSE_TYPE = AlterConfigsResponse_v1
+    SCHEMA = AlterConfigsRequest_v0.SCHEMA
+
+AlterConfigsRequest = [AlterConfigsRequest_v0, AlterConfigsRequest_v1]
+AlterConfigsResponse = [AlterConfigsResponse_v0, AlterConfigsRequest_v1]
 
 
 class DescribeConfigsResponse_v0(Response):
@@ -565,6 +708,28 @@ class DescribeConfigsResponse_v1(Response):
                     ('config_source', Int8)))))))
     )
 
+class DescribeConfigsResponse_v2(Response):
+    API_KEY = 32
+    API_VERSION = 2
+    SCHEMA = Schema(
+        ('throttle_time_ms', Int32),
+        ('resources', Array(
+            ('error_code', Int16),
+            ('error_message', String('utf-8')),
+            ('resource_type', Int8),
+            ('resource_name', String('utf-8')),
+            ('config_entries', Array(
+                ('config_names', String('utf-8')),
+                ('config_value', String('utf-8')),
+                ('read_only', Boolean),
+                ('config_source', Int8),
+                ('is_sensitive', Boolean),
+                ('config_synonyms', Array(
+                    ('config_name', String('utf-8')),
+                    ('config_value', String('utf-8')),
+                    ('config_source', Int8)))))))
+    )
+
 class DescribeConfigsRequest_v0(Request):
     API_KEY = 32
     API_VERSION = 0
@@ -588,10 +753,25 @@ class DescribeConfigsRequest_v1(Request):
         ('include_synonyms', Boolean)
     )
 
-DescribeConfigsRequest = [DescribeConfigsRequest_v0, DescribeConfigsRequest_v1]
-DescribeConfigsResponse = [DescribeConfigsResponse_v0, DescribeConfigsResponse_v1]
 
-class SaslAuthenticateResponse_v0(Request):
+class DescribeConfigsRequest_v2(Request):
+    API_KEY = 32
+    API_VERSION = 2
+    RESPONSE_TYPE = DescribeConfigsResponse_v2
+    SCHEMA = DescribeConfigsRequest_v1.SCHEMA
+
+
+DescribeConfigsRequest = [
+    DescribeConfigsRequest_v0, DescribeConfigsRequest_v1,
+    DescribeConfigsRequest_v2,
+]
+DescribeConfigsResponse = [
+    DescribeConfigsResponse_v0, DescribeConfigsResponse_v1,
+    DescribeConfigsResponse_v2,
+]
+
+
+class SaslAuthenticateResponse_v0(Response):
     API_KEY = 36
     API_VERSION = 0
     SCHEMA = Schema(
@@ -601,6 +781,17 @@ class SaslAuthenticateResponse_v0(Request):
     )
 
 
+class SaslAuthenticateResponse_v1(Response):
+    API_KEY = 36
+    API_VERSION = 1
+    SCHEMA = Schema(
+        ('error_code', Int16),
+        ('error_message', String('utf-8')),
+        ('sasl_auth_bytes', Bytes),
+        ('session_lifetime_ms', Int64)
+    )
+
+
 class SaslAuthenticateRequest_v0(Request):
     API_KEY = 36
     API_VERSION = 0
@@ -610,8 +801,19 @@ class SaslAuthenticateRequest_v0(Request):
     )
 
 
-SaslAuthenticateRequest = [SaslAuthenticateRequest_v0]
-SaslAuthenticateResponse = [SaslAuthenticateResponse_v0]
+class SaslAuthenticateRequest_v1(Request):
+    API_KEY = 36
+    API_VERSION = 1
+    RESPONSE_TYPE = SaslAuthenticateResponse_v1
+    SCHEMA = SaslAuthenticateRequest_v0.SCHEMA
+
+
+SaslAuthenticateRequest = [
+    SaslAuthenticateRequest_v0, SaslAuthenticateRequest_v1,
+]
+SaslAuthenticateResponse = [
+    SaslAuthenticateResponse_v0, SaslAuthenticateResponse_v1,
+]
 
 
 class CreatePartitionsResponse_v0(Response):
@@ -626,6 +828,12 @@ class CreatePartitionsResponse_v0(Response):
     )
 
 
+class CreatePartitionsResponse_v1(Response):
+    API_KEY = 37
+    API_VERSION = 1
+    SCHEMA = CreatePartitionsResponse_v0.SCHEMA
+
+
 class CreatePartitionsRequest_v0(Request):
     API_KEY = 37
     API_VERSION = 0
@@ -641,5 +849,16 @@ class CreatePartitionsRequest_v0(Request):
     )
 
 
-CreatePartitionsRequest = [CreatePartitionsRequest_v0]
-CreatePartitionsResponse = [CreatePartitionsResponse_v0]
+class CreatePartitionsRequest_v1(Request):
+    API_KEY = 37
+    API_VERSION = 1
+    SCHEMA = CreatePartitionsRequest_v0.SCHEMA
+    RESPONSE_TYPE = CreatePartitionsResponse_v1
+
+
+CreatePartitionsRequest = [
+    CreatePartitionsRequest_v0, CreatePartitionsRequest_v1,
+]
+CreatePartitionsResponse = [
+    CreatePartitionsResponse_v0, CreatePartitionsResponse_v1,
+]


Test patch:
diff --git a/test/test_acl_comparisons.py b/test/test_acl_comparisons.py
new file mode 100644
index 0000000..291bf0e
--- /dev/null
+++ b/test/test_acl_comparisons.py
@@ -0,0 +1,92 @@
+from kafka.admin.acl_resource import ACL
+from kafka.admin.acl_resource import ACLOperation
+from kafka.admin.acl_resource import ACLPermissionType
+from kafka.admin.acl_resource import ResourcePattern
+from kafka.admin.acl_resource import ResourceType
+from kafka.admin.acl_resource import ACLResourcePatternType
+
+
+def test_different_acls_are_different():
+    one = ACL(
+        principal='User:A',
+        host='*',
+        operation=ACLOperation.ALL,
+        permission_type=ACLPermissionType.ALLOW,
+        resource_pattern=ResourcePattern(
+            resource_type=ResourceType.TOPIC,
+            resource_name='some-topic',
+            pattern_type=ACLResourcePatternType.LITERAL
+        )
+    )
+
+    two = ACL(
+        principal='User:B',  # Different principal
+        host='*',
+        operation=ACLOperation.ALL,
+        permission_type=ACLPermissionType.ALLOW,
+        resource_pattern=ResourcePattern(
+            resource_type=ResourceType.TOPIC,
+            resource_name='some-topic',
+            pattern_type=ACLResourcePatternType.LITERAL
+        )
+    )
+
+    assert one != two
+    assert hash(one) != hash(two)
+
+def test_different_acls_are_different_with_glob_topics():
+    one = ACL(
+        principal='User:A',
+        host='*',
+        operation=ACLOperation.ALL,
+        permission_type=ACLPermissionType.ALLOW,
+        resource_pattern=ResourcePattern(
+            resource_type=ResourceType.TOPIC,
+            resource_name='*',
+            pattern_type=ACLResourcePatternType.LITERAL
+        )
+    )
+
+    two = ACL(
+        principal='User:B',  # Different principal
+        host='*',
+        operation=ACLOperation.ALL,
+        permission_type=ACLPermissionType.ALLOW,
+        resource_pattern=ResourcePattern(
+            resource_type=ResourceType.TOPIC,
+            resource_name='*',
+            pattern_type=ACLResourcePatternType.LITERAL
+        )
+    )
+
+    assert one != two
+    assert hash(one) != hash(two)
+
+def test_same_acls_are_same():
+    one = ACL(
+        principal='User:A',
+        host='*',
+        operation=ACLOperation.ALL,
+        permission_type=ACLPermissionType.ALLOW,
+        resource_pattern=ResourcePattern(
+            resource_type=ResourceType.TOPIC,
+            resource_name='some-topic',
+            pattern_type=ACLResourcePatternType.LITERAL
+        )
+    )
+
+    two = ACL(
+        principal='User:A',
+        host='*',
+        operation=ACLOperation.ALL,
+        permission_type=ACLPermissionType.ALLOW,
+        resource_pattern=ResourcePattern(
+            resource_type=ResourceType.TOPIC,
+            resource_name='some-topic',
+            pattern_type=ACLResourcePatternType.LITERAL
+        )
+    )
+
+    assert one == two
+    assert hash(one) == hash(two)
+    assert len(set((one, two))) == 1
diff --git a/test/test_api_object_implementation.py b/test/test_api_object_implementation.py
new file mode 100644
index 0000000..da80f14
--- /dev/null
+++ b/test/test_api_object_implementation.py
@@ -0,0 +1,18 @@
+import abc
+import pytest
+
+from kafka.protocol.api import Request
+from kafka.protocol.api import Response
+
+
+attr_names = [n for n in dir(Request) if isinstance(getattr(Request, n), abc.abstractproperty)]
+@pytest.mark.parametrize('klass', Request.__subclasses__())
+@pytest.mark.parametrize('attr_name', attr_names)
+def test_request_type_conformance(klass, attr_name):
+    assert hasattr(klass, attr_name)
+
+attr_names = [n for n in dir(Response) if isinstance(getattr(Response, n), abc.abstractproperty)]
+@pytest.mark.parametrize('klass', Response.__subclasses__())
+@pytest.mark.parametrize('attr_name', attr_names)
+def test_response_type_conformance(klass, attr_name):
+    assert hasattr(klass, attr_name)
diff --git a/test/test_coordinator.py b/test/test_coordinator.py
index 4afdcd9..88ca4c1 100644
--- a/test/test_coordinator.py
+++ b/test/test_coordinator.py
@@ -20,7 +20,7 @@ from kafka.protocol.commit import (
     OffsetCommitRequest, OffsetCommitResponse,
     OffsetFetchRequest, OffsetFetchResponse)
 from kafka.protocol.metadata import MetadataResponse
-from kafka.structs import TopicPartition, OffsetAndMetadata
+from kafka.structs import OffsetAndMetadata, TopicPartition
 from kafka.util import WeakMethod
 
 
@@ -211,7 +211,7 @@ def test_refresh_committed_offsets_if_needed(mocker, coordinator):
     assert coordinator._subscription.needs_fetch_committed_offsets is True
     coordinator.refresh_committed_offsets_if_needed()
     assignment = coordinator._subscription.assignment
-    assert assignment[TopicPartition('foobar', 0)].committed == 123
+    assert assignment[TopicPartition('foobar', 0)].committed == OffsetAndMetadata(123, b'')
     assert TopicPartition('foobar', 1) not in assignment
     assert coordinator._subscription.needs_fetch_committed_offsets is False
 
diff --git a/test/test_fetcher.py b/test/test_fetcher.py
index b61a0f0..697f8be 100644
--- a/test/test_fetcher.py
+++ b/test/test_fetcher.py
@@ -21,7 +21,7 @@ from kafka.errors import (
     UnknownTopicOrPartitionError, OffsetOutOfRangeError
 )
 from kafka.record.memory_records import MemoryRecordsBuilder, MemoryRecords
-from kafka.structs import TopicPartition
+from kafka.structs import OffsetAndMetadata, TopicPartition
 
 
 @pytest.fixture
@@ -124,7 +124,7 @@ def test_update_fetch_positions(fetcher, topic, mocker):
     fetcher._reset_offset.reset_mock()
     fetcher._subscriptions.need_offset_reset(partition)
     fetcher._subscriptions.assignment[partition].awaiting_reset = False
-    fetcher._subscriptions.assignment[partition].committed = 123
+    fetcher._subscriptions.assignment[partition].committed = OffsetAndMetadata(123, b'')
     mocker.patch.object(fetcher._subscriptions, 'seek')
     fetcher.update_fetch_positions([partition])
     assert fetcher._reset_offset.call_count == 0