Repository: dpkp/kafka-python

Issue:
Should consumer.poll() first verify the consumer isn't `closed`?
Is it expected behavior that, depending on the choice of `selector`, you can successfully poll for messages after closing the consumer?

If you call `consumer.close()` followed by `consumer.poll()` that you'll generally (but not always) get an exception... The exception is thrown by the underlying `selector` implementation. `EpollSelector ` and `SelectSelector` throw an error about the file descriptor being closed.

However, I noticed that `PollSelector` does not throw any error, and in a surprising twist, I can continue to consume messages by calling `consumer.poll()` after calling `consumer.close()`. I specifically observed this behavior within a docker container while running some simple tests, as we need to pin the selector due to a `gevent` issue at my day job:

    from kafka import KafkaConsumer, KafkaProducer
    from kafka.client_async import selectors

    bootstrap_server = '172.18.0.6'  # docker IP

    kp = KafkaProducer(bootstrap_servers=[bootstrap_server])

    kp.send('test_topic', 'auto create the topic')
    kp.flush()

    kc = KafkaConsumer('test_topic',
                    group_id='testing-kp-selectors',
                    bootstrap_servers=[bootstrap_server],
                    enable_auto_commit=False,
                    selector=selectors.PollSelector)

    # force the connection to kafka
    p0 = kc.poll(timeout_ms=1000)

    # start the test
    kp.send('test_topic', 't1')
    kp.flush()

    # confirm poll successful
    p1 = kc.poll(timeout_ms=1000)

    assert p1.values()[0][0].value == 't1'

    kc.close()

    # send message after consumer closed
    kp.send('test_topic', 't2')
    kp.flush()

    # continue consuming from closed consumer
    p2 = kc.poll(timeout_ms=1000)
    assert p2.values()[0][0].value == 't2'

    print("Finished")


Should we change this behavior? 

On the one hand, it seems like a natural safeguard that the client should throw an straightforward human-readable error if the connection is closed, rather than an indirect error about an unavailable file descriptor, or, in the case of `PollSelector`, continuing to return records. 

On the other hand, this check would run on every single `poll()` call, and I'm loathe to waste the world's compute energy on something that is only hit in edge cases.

Thoughts?

Discussion:

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/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/group.py b/kafka/consumer/group.py
index 8474b7c..eb7dff2 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -638,6 +638,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 +1174,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/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/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)