Repository: dpkp/kafka-python
Pull request #1972 (2019-12-23 16:56:56)

kafka 2.4.0
adds [kafka 2.4.0](https://www.apache.org/dist/kafka/2.4.0/RELEASE_NOTES.html) to the CI tests

<!-- 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/1972)
<!-- Reviewable:end -->


dpkp: LGTM -- we'll need to merge w/ sasl config changes in #1918

Patch:
diff --git a/.travis.yml b/.travis.yml
index a245650..8e2fdfe 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -14,7 +14,7 @@ env:
     - KAFKA_VERSION=0.10.2.2
     - KAFKA_VERSION=0.11.0.3
     - KAFKA_VERSION=1.1.1
-    - KAFKA_VERSION=2.3.0
+    - KAFKA_VERSION=2.4.0
 
 addons:
   apt:
diff --git a/README.rst b/README.rst
index c4f7f8b..bae567b 100644
--- a/README.rst
+++ b/README.rst
@@ -1,7 +1,7 @@
 Kafka Python client
 ------------------------
 
-.. image:: https://img.shields.io/badge/kafka-2.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
+.. image:: https://img.shields.io/badge/kafka-2.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
     :target: https://kafka-python.readthedocs.io/compatibility.html
 .. image:: https://img.shields.io/pypi/pyversions/kafka-python.svg
     :target: https://pypi.python.org/pypi/kafka-python
@@ -158,4 +158,4 @@ for interacting with kafka brokers via the python repl. This is useful for
 testing, probing, and general experimentation. The protocol support is
 leveraged to enable a KafkaClient.check_version() method that
 probes a kafka broker and attempts to identify which version it is running
-(0.8.0 to 2.3+).
+(0.8.0 to 2.4+).
diff --git a/docs/compatibility.rst b/docs/compatibility.rst
index 9ab877f..93be6fd 100644
--- a/docs/compatibility.rst
+++ b/docs/compatibility.rst
@@ -1,12 +1,12 @@
 Compatibility
 -------------
 
-.. image:: https://img.shields.io/badge/kafka-2.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
+.. image:: https://img.shields.io/badge/kafka-2.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
     :target: https://kafka-python.readthedocs.io/compatibility.html
 .. image:: https://img.shields.io/pypi/pyversions/kafka-python.svg
     :target: https://pypi.python.org/pypi/kafka-python
 
-kafka-python is compatible with (and tested against) broker versions 2.3
+kafka-python is compatible with (and tested against) broker versions 2.4
 through 0.8.0 . kafka-python is not compatible with the 0.8.2-beta release.
 
 Because the kafka server protocol is backwards compatible, kafka-python is
diff --git a/docs/index.rst b/docs/index.rst
index 2322471..fa6f93c 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -1,7 +1,7 @@
 kafka-python
 ############
 
-.. image:: https://img.shields.io/badge/kafka-2.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
+.. image:: https://img.shields.io/badge/kafka-2.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
     :target: https://kafka-python.readthedocs.io/compatibility.html
 .. image:: https://img.shields.io/pypi/pyversions/kafka-python.svg
     :target: https://pypi.python.org/pypi/kafka-python
@@ -136,7 +136,7 @@ for interacting with kafka brokers via the python repl. This is useful for
 testing, probing, and general experimentation. The protocol support is
 leveraged to enable a :meth:`~kafka.KafkaClient.check_version()`
 method that probes a kafka broker and
-attempts to identify which version it is running (0.8.0 to 2.3+).
+attempts to identify which version it is running (0.8.0 to 2.4+).
 
 
 .. toctree::
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,
+]
diff --git a/servers/2.4.0/resources/kafka.properties b/servers/2.4.0/resources/kafka.properties
new file mode 100644
index 0000000..630dbc5
--- /dev/null
+++ b/servers/2.4.0/resources/kafka.properties
@@ -0,0 +1,145 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+# 
+#    http://www.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# see kafka.server.KafkaConfig for additional details and defaults
+
+############################# Server Basics #############################
+
+# The id of the broker. This must be set to a unique integer for each broker.
+broker.id={broker_id}
+
+############################# Socket Server Settings #############################
+
+listeners={transport}://{host}:{port}
+security.inter.broker.protocol={transport}
+
+ssl.keystore.location={ssl_dir}/kafka.server.keystore.jks
+ssl.keystore.password=foobar
+ssl.key.password=foobar
+ssl.truststore.location={ssl_dir}/kafka.server.truststore.jks
+ssl.truststore.password=foobar
+
+authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
+allow.everyone.if.no.acl.found=true
+
+# The port the socket server listens on
+#port=9092
+
+# Hostname the broker will bind to. If not set, the server will bind to all interfaces
+#host.name=localhost
+
+# Hostname the broker will advertise to producers and consumers. If not set, it uses the
+# value for "host.name" if configured.  Otherwise, it will use the value returned from
+# java.net.InetAddress.getCanonicalHostName().
+#advertised.host.name=<hostname routable by clients>
+
+# The port to publish to ZooKeeper for clients to use. If this is not set,
+# it will publish the same port that the broker binds to.
+#advertised.port=<port accessible by clients>
+
+# The number of threads handling network requests
+num.network.threads=3
+ 
+# The number of threads doing disk I/O
+num.io.threads=8
+
+# The send buffer (SO_SNDBUF) used by the socket server
+socket.send.buffer.bytes=102400
+
+# The receive buffer (SO_RCVBUF) used by the socket server
+socket.receive.buffer.bytes=102400
+
+# The maximum size of a request that the socket server will accept (protection against OOM)
+socket.request.max.bytes=104857600
+
+
+############################# Log Basics #############################
+
+# A comma seperated list of directories under which to store log files
+log.dirs={tmp_dir}/data
+
+# The default number of log partitions per topic. More partitions allow greater
+# parallelism for consumption, but this will also result in more files across
+# the brokers.
+num.partitions={partitions}
+default.replication.factor={replicas}
+
+## Short Replica Lag -- Drops failed brokers out of ISR
+replica.lag.time.max.ms=1000
+replica.socket.timeout.ms=1000
+
+############################# Log Flush Policy #############################
+
+# Messages are immediately written to the filesystem but by default we only fsync() to sync
+# the OS cache lazily. The following configurations control the flush of data to disk. 
+# There are a few important trade-offs here:
+#    1. Durability: Unflushed data may be lost if you are not using replication.
+#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
+#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks. 
+# The settings below allow one to configure the flush policy to flush data after a period of time or
+# every N messages (or both). This can be done globally and overridden on a per-topic basis.
+
+# The number of messages to accept before forcing a flush of data to disk
+#log.flush.interval.messages=10000
+
+# The maximum amount of time a message can sit in a log before we force a flush
+#log.flush.interval.ms=1000
+
+############################# Log Retention Policy #############################
+
+# The following configurations control the disposal of log segments. The policy can
+# be set to delete segments after a period of time, or after a given size has accumulated.
+# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
+# from the end of the log.
+
+# The minimum age of a log file to be eligible for deletion
+log.retention.hours=168
+
+# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
+# segments don't drop below log.retention.bytes.
+#log.retention.bytes=1073741824
+
+# The maximum size of a log segment file. When this size is reached a new log segment will be created.
+log.segment.bytes=1073741824
+
+# The interval at which log segments are checked to see if they can be deleted according 
+# to the retention policies
+log.retention.check.interval.ms=300000
+
+# By default the log cleaner is disabled and the log retention policy will default to just delete segments after their retention expires.
+# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.
+log.cleaner.enable=false
+
+# tune down offset topics to reduce setup time in tests
+offsets.commit.timeout.ms=500
+offsets.topic.num.partitions=2
+offsets.topic.replication.factor=1
+
+# Allow shorter session timeouts for tests
+group.min.session.timeout.ms=1000
+
+
+############################# Zookeeper #############################
+
+# Zookeeper connection string (see zookeeper docs for details).
+# This is a comma separated host:port pairs, each corresponding to a zk
+# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
+# You can also append an optional chroot string to the urls to specify the
+# root directory for all kafka znodes.
+zookeeper.connect={zk_host}:{zk_port}/{zk_chroot}
+
+# Timeout in ms for connecting to zookeeper
+zookeeper.connection.timeout.ms=30000
+# We want to expire kafka broker sessions quickly when brokers die b/c we restart them quickly
+zookeeper.session.timeout.ms=500
diff --git a/servers/2.4.0/resources/log4j.properties b/servers/2.4.0/resources/log4j.properties
new file mode 100644
index 0000000..b0b76aa
--- /dev/null
+++ b/servers/2.4.0/resources/log4j.properties
@@ -0,0 +1,25 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+# 
+#    http://www.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+log4j.rootLogger=INFO, stdout, logfile
+
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+log4j.appender.logfile=org.apache.log4j.FileAppender
+log4j.appender.logfile.File=${kafka.logs.dir}/server.log
+log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
+log4j.appender.logfile.layout.ConversionPattern=[%d] %p %m (%c)%n
diff --git a/servers/2.4.0/resources/zookeeper.properties b/servers/2.4.0/resources/zookeeper.properties
new file mode 100644
index 0000000..e3fd097
--- /dev/null
+++ b/servers/2.4.0/resources/zookeeper.properties
@@ -0,0 +1,21 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+# 
+#    http://www.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# the directory where the snapshot is stored.
+dataDir={tmp_dir}
+# the port at which the clients will connect
+clientPort={port}
+clientPortAddress={host}
+# disable the per-ip limit on the number of connections since this is a non-production config
+maxClientCnxns=0


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