Repository: dpkp/kafka-python

Issue:
getting "This socket is already used by another greenlet"
Hi, 
I know you are not testing kafka-python with gevent, anyway it seems work quite good with gevent. 
We currently experiencing only one issue with it. From time to time we getting following exception:

```python
  File "brain/series/writer.py", line 555, in submit
    kafka_async_producer.send('seriesupdates', su)
  File "kafka/producer/kafka.py", line 414, in send
    self._sender.wakeup()
  File "kafka/producer/sender.py", line 293, in wakeup
    self._client.wakeup()
  File "kafka/client_async.py", line 686, in wakeup
    if self._wake_w.send(b'x') != 1:
  File "gevent/_socket2.py", line 325, in send
    self._wait(self._write_event)
  File "gevent/_socket2.py", line 173, in _wait
    raise _socketcommon.ConcurrentObjectUseError('This socket is already used by another greenlet: %r' % (watcher.callback, ))
```

Could you please look into this? Probably some thread lock could solve it?

Thanks!

Discussion:

Patch:
diff --git a/benchmarks/consumer_performance.py b/benchmarks/consumer_performance.py
new file mode 100755
index 0000000..3e879ae
--- /dev/null
+++ b/benchmarks/consumer_performance.py
@@ -0,0 +1,179 @@
+#!/usr/bin/env python
+# Adapted from https://github.com/mrafayaleem/kafka-jython
+
+from __future__ import absolute_import, print_function
+
+import argparse
+import logging
+import pprint
+import sys
+import threading
+import traceback
+
+from kafka import KafkaConsumer, KafkaProducer
+from test.fixtures import KafkaFixture, ZookeeperFixture
+
+logging.basicConfig(level=logging.ERROR)
+
+
+def start_brokers(n):
+    print('Starting {0} {1}-node cluster...'.format(KafkaFixture.kafka_version, n))
+    print('-> 1 Zookeeper')
+    zk = ZookeeperFixture.instance()
+    print('---> {0}:{1}'.format(zk.host, zk.port))
+    print()
+
+    partitions = min(n, 3)
+    replicas = min(n, 3)
+    print('-> {0} Brokers [{1} partitions / {2} replicas]'.format(n, partitions, replicas))
+    brokers = [
+        KafkaFixture.instance(i, zk.host, zk.port, zk_chroot='',
+                              partitions=partitions, replicas=replicas)
+        for i in range(n)
+    ]
+    for broker in brokers:
+        print('---> {0}:{1}'.format(broker.host, broker.port))
+    print()
+    return brokers
+
+
+class ConsumerPerformance(object):
+
+    @staticmethod
+    def run(args):
+        try:
+            props = {}
+            for prop in args.consumer_config:
+                k, v = prop.split('=')
+                try:
+                    v = int(v)
+                except ValueError:
+                    pass
+                if v == 'None':
+                    v = None
+                props[k] = v
+
+            if args.brokers:
+                brokers = start_brokers(args.brokers)
+                props['bootstrap_servers'] = ['{0}:{1}'.format(broker.host, broker.port)
+                                              for broker in brokers]
+                print('---> bootstrap_servers={0}'.format(props['bootstrap_servers']))
+                print()
+
+                print('-> Producing records')
+                record = bytes(bytearray(args.record_size))
+                producer = KafkaProducer(compression_type=args.fixture_compression,
+                                         **props)
+                for i in xrange(args.num_records):
+                    producer.send(topic=args.topic, value=record)
+                producer.flush()
+                producer.close()
+                print('-> OK!')
+                print()
+
+            print('Initializing Consumer...')
+            props['auto_offset_reset'] = 'earliest'
+            if 'consumer_timeout_ms' not in props:
+                props['consumer_timeout_ms'] = 10000
+            props['metrics_sample_window_ms'] = args.stats_interval * 1000
+            for k, v in props.items():
+                print('---> {0}={1}'.format(k, v))
+            consumer = KafkaConsumer(args.topic, **props)
+            print('---> group_id={0}'.format(consumer.config['group_id']))
+            print('---> report stats every {0} secs'.format(args.stats_interval))
+            print('---> raw metrics? {0}'.format(args.raw_metrics))
+            timer_stop = threading.Event()
+            timer = StatsReporter(args.stats_interval, consumer,
+                                  event=timer_stop,
+                                  raw_metrics=args.raw_metrics)
+            timer.start()
+            print('-> OK!')
+            print()
+
+            records = 0
+            for msg in consumer:
+                records += 1
+                if records >= args.num_records:
+                    break
+            print('Consumed {0} records'.format(records))
+
+            timer_stop.set()
+
+        except Exception:
+            exc_info = sys.exc_info()
+            traceback.print_exception(*exc_info)
+            sys.exit(1)
+
+
+class StatsReporter(threading.Thread):
+    def __init__(self, interval, consumer, event=None, raw_metrics=False):
+        super(StatsReporter, self).__init__()
+        self.interval = interval
+        self.consumer = consumer
+        self.event = event
+        self.raw_metrics = raw_metrics
+
+    def print_stats(self):
+        metrics = self.consumer.metrics()
+        if self.raw_metrics:
+            pprint.pprint(metrics)
+        else:
+            print('{records-consumed-rate} records/sec ({bytes-consumed-rate} B/sec),'
+                  ' {fetch-latency-avg} latency,'
+                  ' {fetch-rate} fetch/s,'
+                  ' {fetch-size-avg} fetch size,'
+                  ' {records-lag-max} max record lag,'
+                  ' {records-per-request-avg} records/req'
+                  .format(**metrics['consumer-fetch-manager-metrics']))
+
+
+    def print_final(self):
+        self.print_stats()
+
+    def run(self):
+        while self.event and not self.event.wait(self.interval):
+            self.print_stats()
+        else:
+            self.print_final()
+
+
+def get_args_parser():
+    parser = argparse.ArgumentParser(
+        description='This tool is used to verify the consumer performance.')
+
+    parser.add_argument(
+        '--topic', type=str,
+        help='Topic for consumer test',
+        default='kafka-python-benchmark-test')
+    parser.add_argument(
+        '--num-records', type=long,
+        help='number of messages to consume',
+        default=1000000)
+    parser.add_argument(
+        '--record-size', type=int,
+        help='message size in bytes',
+        default=100)
+    parser.add_argument(
+        '--consumer-config', type=str, nargs='+', default=(),
+        help='kafka consumer related configuaration properties like '
+             'bootstrap_servers,client_id etc..')
+    parser.add_argument(
+        '--fixture-compression', type=str,
+        help='specify a compression type for use with broker fixtures / producer')
+    parser.add_argument(
+        '--brokers', type=int,
+        help='Number of kafka brokers to start',
+        default=0)
+    parser.add_argument(
+        '--stats-interval', type=int,
+        help='Interval in seconds for stats reporting to console',
+        default=5)
+    parser.add_argument(
+        '--raw-metrics', action='store_true',
+        help='Enable this flag to print full metrics dict on each interval')
+    return parser
+
+
+if __name__ == '__main__':
+    args = get_args_parser().parse_args()
+    ConsumerPerformance.run(args)
diff --git a/load_example.py b/benchmarks/load_example.py
similarity index 100%
rename from load_example.py
rename to benchmarks/load_example.py
diff --git a/benchmarks/producer_performance.py b/benchmarks/producer_performance.py
new file mode 100755
index 0000000..e958735
--- /dev/null
+++ b/benchmarks/producer_performance.py
@@ -0,0 +1,158 @@
+#!/usr/bin/env python
+# Adapted from https://github.com/mrafayaleem/kafka-jython
+
+from __future__ import absolute_import, print_function
+
+import argparse
+import pprint
+import sys
+import threading
+import traceback
+
+from kafka import KafkaProducer
+from test.fixtures import KafkaFixture, ZookeeperFixture
+
+
+def start_brokers(n):
+    print('Starting {0} {1}-node cluster...'.format(KafkaFixture.kafka_version, n))
+    print('-> 1 Zookeeper')
+    zk = ZookeeperFixture.instance()
+    print('---> {0}:{1}'.format(zk.host, zk.port))
+    print()
+
+    partitions = min(n, 3)
+    replicas = min(n, 3)
+    print('-> {0} Brokers [{1} partitions / {2} replicas]'.format(n, partitions, replicas))
+    brokers = [
+        KafkaFixture.instance(i, zk.host, zk.port, zk_chroot='',
+                              partitions=partitions, replicas=replicas)
+        for i in range(n)
+    ]
+    for broker in brokers:
+        print('---> {0}:{1}'.format(broker.host, broker.port))
+    print()
+    return brokers
+
+
+class ProducerPerformance(object):
+
+    @staticmethod
+    def run(args):
+        try:
+            props = {}
+            for prop in args.producer_config:
+                k, v = prop.split('=')
+                try:
+                    v = int(v)
+                except ValueError:
+                    pass
+                if v == 'None':
+                    v = None
+                props[k] = v
+
+            if args.brokers:
+                brokers = start_brokers(args.brokers)
+                props['bootstrap_servers'] = ['{0}:{1}'.format(broker.host, broker.port)
+                                              for broker in brokers]
+                print("---> bootstrap_servers={0}".format(props['bootstrap_servers']))
+                print()
+                print('-> OK!')
+                print()
+
+            print('Initializing producer...')
+            record = bytes(bytearray(args.record_size))
+            props['metrics_sample_window_ms'] = args.stats_interval * 1000
+
+            producer = KafkaProducer(**props)
+            for k, v in props.items():
+                print('---> {0}={1}'.format(k, v))
+            print('---> send {0} byte records'.format(args.record_size))
+            print('---> report stats every {0} secs'.format(args.stats_interval))
+            print('---> raw metrics? {0}'.format(args.raw_metrics))
+            timer_stop = threading.Event()
+            timer = StatsReporter(args.stats_interval, producer,
+                                  event=timer_stop,
+                                  raw_metrics=args.raw_metrics)
+            timer.start()
+            print('-> OK!')
+            print()
+
+            for i in xrange(args.num_records):
+                producer.send(topic=args.topic, value=record)
+            producer.flush()
+
+            timer_stop.set()
+
+        except Exception:
+            exc_info = sys.exc_info()
+            traceback.print_exception(*exc_info)
+            sys.exit(1)
+
+
+class StatsReporter(threading.Thread):
+    def __init__(self, interval, producer, event=None, raw_metrics=False):
+        super(StatsReporter, self).__init__()
+        self.interval = interval
+        self.producer = producer
+        self.event = event
+        self.raw_metrics = raw_metrics
+
+    def print_stats(self):
+        metrics = self.producer.metrics()
+        if self.raw_metrics:
+            pprint.pprint(metrics)
+        else:
+            print('{record-send-rate} records/sec ({byte-rate} B/sec),'
+                  ' {request-latency-avg} latency,'
+                  ' {record-size-avg} record size,'
+                  ' {batch-size-avg} batch size,'
+                  ' {records-per-request-avg} records/req'
+                  .format(**metrics['producer-metrics']))
+
+    def print_final(self):
+        self.print_stats()
+
+    def run(self):
+        while self.event and not self.event.wait(self.interval):
+            self.print_stats()
+        else:
+            self.print_final()
+
+
+def get_args_parser():
+    parser = argparse.ArgumentParser(
+        description='This tool is used to verify the producer performance.')
+
+    parser.add_argument(
+        '--topic', type=str,
+        help='Topic name for test',
+        default='kafka-python-benchmark-test')
+    parser.add_argument(
+        '--num-records', type=long,
+        help='number of messages to produce',
+        default=1000000)
+    parser.add_argument(
+        '--record-size', type=int,
+        help='message size in bytes',
+        default=100)
+    parser.add_argument(
+        '--producer-config', type=str, nargs='+', default=(),
+        help='kafka producer related configuaration properties like '
+             'bootstrap_servers,client_id etc..')
+    parser.add_argument(
+        '--brokers', type=int,
+        help='Number of kafka brokers to start',
+        default=0)
+    parser.add_argument(
+        '--stats-interval', type=int,
+        help='Interval in seconds for stats reporting to console',
+        default=5)
+    parser.add_argument(
+        '--raw-metrics', action='store_true',
+        help='Enable this flag to print full metrics dict on each interval')
+    return parser
+
+
+if __name__ == '__main__':
+    args = get_args_parser().parse_args()
+    ProducerPerformance.run(args)
diff --git a/kafka/client.py b/kafka/client.py
index 891ae03..8a34cc4 100644
--- a/kafka/client.py
+++ b/kafka/client.py
@@ -137,7 +137,7 @@ class SimpleClient(object):
         kafka.errors.check_error(resp)
 
         # Otherwise return the BrokerMetadata
-        return BrokerMetadata(resp.nodeId, resp.host, resp.port)
+        return BrokerMetadata(resp.nodeId, resp.host, resp.port, None)
 
     def _next_id(self):
         """Generate a new correlation id"""
@@ -525,7 +525,7 @@ class SimpleClient(object):
         log.debug('Updating broker metadata: %s', resp.brokers)
         log.debug('Updating topic metadata: %s', [topic for _, topic, _ in resp.topics])
 
-        self.brokers = dict([(nodeId, BrokerMetadata(nodeId, host, port))
+        self.brokers = dict([(nodeId, BrokerMetadata(nodeId, host, port, None))
                              for nodeId, host, port in resp.brokers])
 
         for error, topic, partitions in resp.topics:
@@ -577,7 +577,7 @@ class SimpleClient(object):
                 # (not sure how this could happen. server could be in bad state)
                 else:
                     self.topics_to_brokers[topic_part] = BrokerMetadata(
-                        leader, None, None
+                        leader, None, None, None
                     )
 
     def send_metadata_request(self, payloads=[], fail_on_error=True,
diff --git a/kafka/client_async.py b/kafka/client_async.py
index 6fa9434..2700069 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -6,6 +6,7 @@ import heapq
 import itertools
 import logging
 import random
+import threading
 
 # selectors in stdlib as of py3.4
 try:
@@ -158,6 +159,7 @@ class KafkaClient(object):
         self._bootstrap_fails = 0
         self._wake_r, self._wake_w = socket.socketpair()
         self._wake_r.setblocking(False)
+        self._wake_lock = threading.Lock()
         self._selector.register(self._wake_r, selectors.EVENT_READ)
         self._closed = False
         self._bootstrap(collect_hosts(self.config['bootstrap_servers']))
@@ -178,7 +180,11 @@ class KafkaClient(object):
             time.sleep(next_at - now)
         self._last_bootstrap = time.time()
 
-        metadata_request = MetadataRequest[0]([])
+        if self.config['api_version'] is None or self.config['api_version'] < (0, 10):
+            metadata_request = MetadataRequest[0]([])
+        else:
+            metadata_request = MetadataRequest[1](None)
+
         for host, port, afi in hosts:
             log.debug("Attempting to bootstrap via node at %s:%s", host, port)
             cb = functools.partial(self._conn_state_change, 'bootstrap')
@@ -643,10 +649,17 @@ class KafkaClient(object):
 
             topics = list(self._topics)
             if self.cluster.need_all_topic_metadata:
-                topics = []
+                if self.config['api_version'] < (0, 10):
+                    topics = []
+                else:
+                    topics = None
 
             if self._can_send_request(node_id):
-                request = MetadataRequest[0](topics)
+                if self.config['api_version'] < (0, 10):
+                    api_version = 0
+                else:
+                    api_version = 1
+                request = MetadataRequest[api_version](topics)
                 log.debug("Sending metadata request %s to node %s", request, node_id)
                 future = self.send(node_id, request)
                 future.add_callback(self.cluster.update_metadata)
@@ -747,10 +760,12 @@ class KafkaClient(object):
             raise Errors.NoBrokersAvailable()
 
     def wakeup(self):
-        if self._wake_w.send(b'x') != 1:
-            log.warning('Unable to send to wakeup socket!')
+        with self._wake_lock:
+            if self._wake_w.send(b'x') != 1:
+                log.warning('Unable to send to wakeup socket!')
 
     def _clear_wake_fd(self):
+        # reading from wake socket should only happen in a single thread
         while True:
             try:
                 self._wake_r.recv(1024)
diff --git a/kafka/cluster.py b/kafka/cluster.py
index 9aabec1..694e115 100644
--- a/kafka/cluster.py
+++ b/kafka/cluster.py
@@ -34,6 +34,8 @@ class ClusterMetadata(object):
         self._lock = threading.Lock()
         self.need_all_topic_metadata = False
         self.unauthorized_topics = set()
+        self.internal_topics = set()
+        self.controller = None
 
         self.config = copy.copy(self.DEFAULT_CONFIG)
         for key in self.config:
@@ -150,13 +152,23 @@ class ClusterMetadata(object):
               self._future = Future()
             return self._future
 
-    def topics(self):
+    def topics(self, exclude_internal_topics=True):
         """Get set of known topics.
 
+        Arguments:
+            exclude_internal_topics (bool): Whether records from internal topics
+                (such as offsets) should be exposed to the consumer. If set to
+                True the only way to receive records from an internal topic is
+                subscribing to it. Default True
+
         Returns:
             set: {topic (str), ...}
         """
-        return set(self._partitions.keys())
+        topics = set(self._partitions.keys())
+        if exclude_internal_topics:
+            return topics - self.internal_topics
+        else:
+            return topics
 
     def failed_update(self, exception):
         """Update cluster state given a failed MetadataRequest."""
@@ -180,23 +192,41 @@ class ClusterMetadata(object):
         # In the common case where we ask for a single topic and get back an
         # error, we should fail the future
         if len(metadata.topics) == 1 and metadata.topics[0][0] != 0:
-            error_code, topic, _ = metadata.topics[0]
+            error_code, topic = metadata.topics[0][:2]
             error = Errors.for_code(error_code)(topic)
             return self.failed_update(error)
 
         if not metadata.brokers:
             log.warning("No broker metadata found in MetadataResponse")
 
-        for node_id, host, port in metadata.brokers:
+        for broker in metadata.brokers:
+            if metadata.API_VERSION == 0:
+                node_id, host, port = broker
+                rack = None
+            else:
+                node_id, host, port, rack = broker
             self._brokers.update({
-                node_id: BrokerMetadata(node_id, host, port)
+                node_id: BrokerMetadata(node_id, host, port, rack)
             })
 
+        if metadata.API_VERSION == 0:
+            self.controller = None
+        else:
+            self.controller = self._brokers.get(metadata.controller_id)
+
         _new_partitions = {}
         _new_broker_partitions = collections.defaultdict(set)
         _new_unauthorized_topics = set()
+        _new_internal_topics = set()
 
-        for error_code, topic, partitions in metadata.topics:
+        for topic_data in metadata.topics:
+            if metadata.API_VERSION == 0:
+                error_code, topic, partitions = topic_data
+                is_internal = False
+            else:
+                error_code, topic, is_internal, partitions = topic_data
+            if is_internal:
+                _new_internal_topics.add(topic)
             error_type = Errors.for_code(error_code)
             if error_type is Errors.NoError:
                 _new_partitions[topic] = {}
@@ -226,6 +256,7 @@ class ClusterMetadata(object):
             self._partitions = _new_partitions
             self._broker_partitions = _new_broker_partitions
             self.unauthorized_topics = _new_unauthorized_topics
+            self.internal_topics = _new_internal_topics
             f = None
             if self._future:
                 f = self._future
@@ -272,7 +303,8 @@ class ClusterMetadata(object):
         coordinator = BrokerMetadata(
             response.coordinator_id,
             response.host,
-            response.port)
+            response.port,
+            None)
 
         # Assume that group coordinators are just brokers
         # (this is true now, but could diverge in future)
@@ -281,12 +313,14 @@ class ClusterMetadata(object):
 
         # If this happens, either brokers have moved without
         # changing IDs, or our assumption above is wrong
-        elif coordinator != self._brokers[node_id]:
-            log.error("GroupCoordinator metadata conflicts with existing"
-                      " broker metadata. Coordinator: %s, Broker: %s",
-                      coordinator, self._brokers[node_id])
-            self._groups[group] = node_id
-            return False
+        else:
+            node = self._brokers[node_id]
+            if coordinator.host != node.host or coordinator.port != node.port:
+                log.error("GroupCoordinator metadata conflicts with existing"
+                          " broker metadata. Coordinator: %s, Broker: %s",
+                          coordinator, node)
+                self._groups[group] = node_id
+                return False
 
         log.info("Group coordinator for %s is %s", group, coordinator)
         self._groups[group] = node_id
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py
index 34ff4cb..d615848 100644
--- a/kafka/consumer/fetcher.py
+++ b/kafka/consumer/fetcher.py
@@ -729,6 +729,8 @@ class Fetcher(six.Iterator):
                 else:
                     raise error_type('Unexpected error while fetching data')
 
+        # Because we are currently decompressing messages lazily, the sensors here
+        # will get compressed bytes / message set stats when compression is enabled
         self._sensors.bytes_fetched.record(total_bytes)
         self._sensors.records_fetched.record(total_count)
         if response.API_VERSION >= 1:
@@ -774,12 +776,12 @@ class FetchManagerMetrics(object):
             'The maximum throttle time in ms'), Max())
 
     def record_topic_fetch_metrics(self, topic, num_bytes, num_records):
-        metric_tags = {'topic': topic.replace('.', '_')}
-
         # record bytes fetched
         name = '.'.join(['topic', topic, 'bytes-fetched'])
         bytes_fetched = self.metrics.get_sensor(name)
         if not bytes_fetched:
+            metric_tags = {'topic': topic.replace('.', '_')}
+
             bytes_fetched = self.metrics.sensor(name)
             bytes_fetched.add(self.metrics.metric_name('fetch-size-avg',
                     self.group_name,
@@ -799,6 +801,8 @@ class FetchManagerMetrics(object):
         name = '.'.join(['topic', topic, 'records-fetched'])
         records_fetched = self.metrics.get_sensor(name)
         if not records_fetched:
+            metric_tags = {'topic': topic.replace('.', '_')}
+
             records_fetched = self.metrics.sensor(name)
             records_fetched.add(self.metrics.metric_name('records-per-request-avg',
                     self.group_name,
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index 8fa43bc..b2114d8 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -12,7 +12,7 @@ from kafka.consumer.subscription_state import SubscriptionState
 from kafka.coordinator.consumer import ConsumerCoordinator
 from kafka.coordinator.assignors.range import RangePartitionAssignor
 from kafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor
-from kafka.metrics import DictReporter, MetricConfig, Metrics
+from kafka.metrics import MetricConfig, Metrics
 from kafka.protocol.offset import OffsetResetStrategy
 from kafka.structs import TopicPartition
 from kafka.version import __version__
@@ -171,12 +171,12 @@ class KafkaConsumer(six.Iterator):
             in classes that will be notified of new metric creation. Default: []
         metrics_num_samples (int): The number of samples maintained to compute
             metrics. Default: 2
-        metrics_sample_window_ms (int): The number of samples maintained to
-            compute metrics. Default: 30000
+        metrics_sample_window_ms (int): The maximum age in milliseconds of
+            samples used to compute metrics. Default: 30000
 
     Note:
         Configuration parameters are described in more detail at
-        https://kafka.apache.org/090/configuration.html#newconsumerconfigs
+        https://kafka.apache.org/0100/configuration.html#newconsumerconfigs
     """
     DEFAULT_CONFIG = {
         'bootstrap_servers': 'localhost',
@@ -241,7 +241,6 @@ class KafkaConsumer(six.Iterator):
                                      time_window_ms=self.config['metrics_sample_window_ms'],
                                      tags=metrics_tags)
         reporters = [reporter() for reporter in self.config['metric_reporters']]
-        reporters.append(DictReporter('kafka.consumer'))
         self._metrics = Metrics(metric_config, reporters)
         metric_group_prefix = 'consumer'
         # TODO _metrics likely needs to be passed to KafkaClient, etc.
@@ -760,6 +759,21 @@ class KafkaConsumer(six.Iterator):
         self._client.set_topics([])
         log.debug("Unsubscribed all topics or patterns and assigned partitions")
 
+    def metrics(self, raw=False):
+        """Warning: this is an unstable interface.
+        It may change in future releases without warning"""
+        if raw:
+            return self._metrics.metrics
+
+        metrics = {}
+        for k, v in self._metrics.metrics.items():
+            if k.group not in metrics:
+                metrics[k.group] = {}
+            if k.name not in metrics[k.group]:
+                metrics[k.group][k.name] = {}
+            metrics[k.group][k.name] = v.value()
+        return metrics
+
     def _use_consumer_group(self):
         """Return True iff this consumer can/should join a broker-coordinated group."""
         if self.config['api_version'] < (0, 9):
diff --git a/kafka/metrics/stats/sensor.py b/kafka/metrics/stats/sensor.py
index b0bf4db..72bacfc 100644
--- a/kafka/metrics/stats/sensor.py
+++ b/kafka/metrics/stats/sensor.py
@@ -55,15 +55,15 @@ class Sensor(object):
         Record a value at a known time.
         Arguments:
             value (double): The value we are recording
-            time_ms (int): The current POSIX time in milliseconds
+            time_ms (int): A POSIX timestamp in milliseconds.
+                Default: The time when record() is evaluated (now)
 
         Raises:
             QuotaViolationException: if recording this value moves a
                 metric beyond its configured maximum or minimum bound
         """
-        now = time.time() * 1000
         if time_ms is None:
-            time_ms = now
+            time_ms = time.time() * 1000
         self._last_record_time = time_ms
         with self._lock:  # XXX high volume, might be performance issue
             # increment all the stats
diff --git a/kafka/producer/buffer.py b/kafka/producer/buffer.py
index 5fcb35f..de5f0e7 100644
--- a/kafka/producer/buffer.py
+++ b/kafka/producer/buffer.py
@@ -1,4 +1,4 @@
-from __future__ import absolute_import
+from __future__ import absolute_import, division
 
 import collections
 import io
@@ -55,6 +55,8 @@ class MessageSetBuffer(object):
         self._batch_size = batch_size
         self._closed = False
         self._messages = 0
+        self._bytes_written = 4 # Int32 header is 4 bytes
+        self._final_size = None
 
     def append(self, offset, message):
         """Apend a Message to the MessageSet.
@@ -62,6 +64,8 @@ class MessageSetBuffer(object):
         Arguments:
             offset (int): offset of the message
             message (Message or bytes): message struct or encoded bytes
+
+        Returns: bytes written
         """
         if isinstance(message, Message):
             encoded = message.encode()
@@ -70,6 +74,8 @@ class MessageSetBuffer(object):
         msg = Int64.encode(offset) + Int32.encode(len(encoded)) + encoded
         self._buffer.write(msg)
         self._messages += 1
+        self._bytes_written += len(msg)
+        return len(msg)
 
     def has_room_for(self, key, value):
         if self._closed:
@@ -107,16 +113,20 @@ class MessageSetBuffer(object):
                 self._buffer.write(Int32.encode(len(encoded)))
                 self._buffer.write(encoded)
 
-            # Update the message set size, and return ready for full read()
-            size = self._buffer.tell() - 4
+            # Update the message set size (less the 4 byte header),
+            # and return with buffer ready for full read()
+            self._final_size = self._buffer.tell()
             self._buffer.seek(0)
-            self._buffer.write(Int32.encode(size))
+            self._buffer.write(Int32.encode(self._final_size - 4))
 
         self._buffer.seek(0)
         self._closed = True
 
     def size_in_bytes(self):
-        return self._buffer.tell()
+        return self._final_size or self._buffer.tell()
+
+    def compression_rate(self):
+        return self.size_in_bytes() / self._bytes_written
 
     def buffer(self):
         return self._buffer
diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
index f5c5d19..af07154 100644
--- a/kafka/producer/kafka.py
+++ b/kafka/producer/kafka.py
@@ -9,6 +9,7 @@ import weakref
 
 from .. import errors as Errors
 from ..client_async import KafkaClient
+from ..metrics import MetricConfig, Metrics
 from ..partitioner.default import DefaultPartitioner
 from ..protocol.message import Message, MessageSet
 from ..structs import TopicPartition
@@ -220,10 +221,17 @@ class KafkaProducer(object):
         api_version_auto_timeout_ms (int): number of milliseconds to throw a
             timeout exception from the constructor when checking the broker
             api version. Only applies if api_version set to 'auto'
+        metric_reporters (list): A list of classes to use as metrics reporters.
+            Implementing the AbstractMetricsReporter interface allows plugging
+            in classes that will be notified of new metric creation. Default: []
+        metrics_num_samples (int): The number of samples maintained to compute
+            metrics. Default: 2
+        metrics_sample_window_ms (int): The maximum age in milliseconds of
+            samples used to compute metrics. Default: 30000
 
     Note:
         Configuration parameters are described in more detail at
-        https://kafka.apache.org/090/configuration.html#producerconfigs
+        https://kafka.apache.org/0100/configuration.html#producerconfigs
     """
     _DEFAULT_CONFIG = {
         'bootstrap_servers': 'localhost',
@@ -255,7 +263,10 @@ class KafkaProducer(object):
         'ssl_keyfile': None,
         'ssl_crlfile': None,
         'api_version': None,
-        'api_version_auto_timeout_ms': 2000
+        'api_version_auto_timeout_ms': 2000,
+        'metric_reporters': [],
+        'metrics_num_samples': 2,
+        'metrics_sample_window_ms': 30000,
     }
 
     def __init__(self, **configs):
@@ -285,6 +296,14 @@ class KafkaProducer(object):
             log.warning('use api_version=%s (%s is deprecated)',
                         str(self.config['api_version']), deprecated)
 
+        # Configure metrics
+        metrics_tags = {'client-id': self.config['client_id']}
+        metric_config = MetricConfig(samples=self.config['metrics_num_samples'],
+                                     time_window_ms=self.config['metrics_sample_window_ms'],
+                                     tags=metrics_tags)
+        reporters = [reporter() for reporter in self.config['metric_reporters']]
+        self._metrics = Metrics(metric_config, reporters)
+
         client = KafkaClient(**self.config)
 
         # Get auto-discovered version from client if necessary
@@ -298,7 +317,8 @@ class KafkaProducer(object):
         self._accumulator = RecordAccumulator(message_version=message_version, **self.config)
         self._metadata = client.cluster
         guarantee_message_order = bool(self.config['max_in_flight_requests_per_connection'] == 1)
-        self._sender = Sender(client, self._metadata, self._accumulator,
+        self._sender = Sender(client, self._metadata,
+                              self._accumulator, self._metrics,
                               guarantee_message_order=guarantee_message_order,
                               **self.config)
         self._sender.daemon = True
@@ -382,6 +402,7 @@ class KafkaProducer(object):
             if not invoked_from_callback:
                 self._sender.join()
 
+        self._metrics.close()
         try:
             self.config['key_serializer'].close()
         except AttributeError:
@@ -581,3 +602,18 @@ class KafkaProducer(object):
         return self.config['partitioner'](serialized_key,
                                           all_partitions,
                                           available)
+
+    def metrics(self, raw=False):
+        """Warning: this is an unstable interface.
+        It may change in future releases without warning"""
+        if raw:
+            return self._metrics.metrics
+
+        metrics = {}
+        for k, v in self._metrics.metrics.items():
+            if k.group not in metrics:
+                metrics[k.group] = {}
+            if k.name not in metrics[k.group]:
+                metrics[k.group][k.name] = {}
+            metrics[k.group][k.name] = v.value()
+        return metrics
diff --git a/kafka/producer/record_accumulator.py b/kafka/producer/record_accumulator.py
index 566bf6f..7ea579a 100644
--- a/kafka/producer/record_accumulator.py
+++ b/kafka/producer/record_accumulator.py
@@ -38,7 +38,7 @@ class AtomicInteger(object):
 class RecordBatch(object):
     def __init__(self, tp, records, message_version=0):
         self.record_count = 0
-        #self.max_record_size = 0 # for metrics only
+        self.max_record_size = 0
         now = time.time()
         self.created = now
         self.drained = None
@@ -56,8 +56,8 @@ class RecordBatch(object):
             return None
 
         msg = Message(value, key=key, magic=self.message_version)
-        self.records.append(self.record_count, msg)
-        # self.max_record_size = max(self.max_record_size, Record.record_size(key, value)) # for metrics only
+        record_size = self.records.append(self.record_count, msg)
+        self.max_record_size = max(self.max_record_size, record_size)
         self.last_append = time.time()
         future = FutureRecordMetadata(self.produce_future, self.record_count,
                                       timestamp_ms)
diff --git a/kafka/producer/sender.py b/kafka/producer/sender.py
index 958e165..c1d0905 100644
--- a/kafka/producer/sender.py
+++ b/kafka/producer/sender.py
@@ -1,4 +1,4 @@
-from __future__ import absolute_import
+from __future__ import absolute_import, division
 
 import collections
 import copy
@@ -8,9 +8,11 @@ import threading
 import six
 
 from .. import errors as Errors
+from ..metrics.measurable import AnonMeasurable
+from ..metrics.stats import Avg, Count, Max, Rate
+from ..protocol.produce import ProduceRequest
 from ..structs import TopicPartition
 from ..version import __version__
-from ..protocol.produce import ProduceRequest
 
 log = logging.getLogger(__name__)
 
@@ -31,7 +33,7 @@ class Sender(threading.Thread):
         'api_version': (0, 8, 0),
     }
 
-    def __init__(self, client, metadata, accumulator, **configs):
+    def __init__(self, client, metadata, accumulator, metrics, **configs):
         super(Sender, self).__init__()
         self.config = copy.copy(self._DEFAULT_CONFIG)
         for key in self.config:
@@ -45,6 +47,7 @@ class Sender(threading.Thread):
         self._running = True
         self._force_close = False
         self._topics_to_add = set()
+        self._sensors = SenderMetrics(metrics, self._client, self._metadata)
 
     def run(self):
         """The main run loop for the sender thread."""
@@ -119,7 +122,10 @@ class Sender(threading.Thread):
 
         expired_batches = self._accumulator.abort_expired_batches(
             self.config['request_timeout_ms'], self._metadata)
+        for expired_batch in expired_batches:
+            self._sensors.record_errors(expired_batch.topic_partition.topic, expired_batch.record_count)
 
+        self._sensors.update_produce_request_metrics(batches_by_node)
         requests = self._create_produce_requests(batches_by_node)
         # If we have any nodes that are ready to send + have sendable data,
         # poll with 0 timeout so this can immediately loop and try sending more
@@ -223,6 +229,7 @@ class Sender(threading.Thread):
                         self.config['retries'] - batch.attempts - 1,
                         error)
             self._accumulator.reenqueue(batch)
+            self._sensors.record_retries(batch.topic_partition.topic, batch.record_count)
         else:
             if error is Errors.TopicAuthorizationFailedError:
                 error = error(batch.topic_partition.topic)
@@ -230,6 +237,8 @@ class Sender(threading.Thread):
             # tell the user the result of their request
             batch.done(base_offset, timestamp_ms, error)
             self._accumulator.deallocate(batch)
+            if error is not None:
+                self._sensors.record_errors(batch.topic_partition.topic, batch.record_count)
 
         if getattr(error, 'invalid_metadata', False):
             self._metadata.request_update()
@@ -296,3 +305,200 @@ class Sender(threading.Thread):
     def wakeup(self):
         """Wake up the selector associated with this send thread."""
         self._client.wakeup()
+
+
+class SenderMetrics(object):
+
+    def __init__(self, metrics, client, metadata):
+        self.metrics = metrics
+        self._client = client
+        self._metadata = metadata
+
+        sensor_name = 'batch-size'
+        self.batch_size_sensor = self.metrics.sensor(sensor_name)
+        self.add_metric('batch-size-avg', Avg(),
+                        sensor_name=sensor_name,
+                        description='The average number of bytes sent per partition per-request.')
+        self.add_metric('batch-size-max', Max(),
+                        sensor_name=sensor_name,
+                        description='The max number of bytes sent per partition per-request.')
+
+        sensor_name = 'compression-rate'
+        self.compression_rate_sensor = self.metrics.sensor(sensor_name)
+        self.add_metric('compression-rate-avg', Avg(),
+                        sensor_name=sensor_name,
+                        description='The average compression rate of record batches.')
+
+        sensor_name = 'queue-time'
+        self.queue_time_sensor = self.metrics.sensor(sensor_name)
+        self.add_metric('record-queue-time-avg', Avg(),
+                        sensor_name=sensor_name,
+                        description='The average time in ms record batches spent in the record accumulator.')
+        self.add_metric('record-queue-time-max', Max(),
+                        sensor_name=sensor_name,
+                        description='The maximum time in ms record batches spent in the record accumulator.')
+
+        sensor_name = 'request-time'
+        self.request_time_sensor = self.metrics.sensor(sensor_name)
+        self.add_metric('request-latency-avg', Avg(),
+                        sensor_name=sensor_name,
+                        description='The average request latency in ms')
+        self.add_metric('request-latency-max', Max(),
+                        sensor_name=sensor_name,
+                        description='The maximum request latency in ms')
+
+        sensor_name = 'produce-throttle-time'
+        self.produce_throttle_time_sensor = self.metrics.sensor(sensor_name)
+        self.add_metric('produce-throttle-time-avg', Avg(),
+                        sensor_name=sensor_name,
+                        description='The average throttle time in ms')
+        self.add_metric('produce-throttle-time-max', Max(),
+                        sensor_name=sensor_name,
+                        description='The maximum throttle time in ms')
+
+        sensor_name = 'records-per-request'
+        self.records_per_request_sensor = self.metrics.sensor(sensor_name)
+        self.add_metric('record-send-rate', Rate(),
+                        sensor_name=sensor_name,
+                        description='The average number of records sent per second.')
+        self.add_metric('records-per-request-avg', Avg(),
+                        sensor_name=sensor_name,
+                        description='The average number of records per request.')
+
+        sensor_name = 'bytes'
+        self.byte_rate_sensor = self.metrics.sensor(sensor_name)
+        self.add_metric('byte-rate', Rate(),
+                        sensor_name=sensor_name,
+                        description='The average number of bytes sent per second.')
+
+        sensor_name = 'record-retries'
+        self.retry_sensor = self.metrics.sensor(sensor_name)
+        self.add_metric('record-retry-rate', Rate(),
+                        sensor_name=sensor_name,
+                        description='The average per-second number of retried record sends')
+
+        sensor_name = 'errors'
+        self.error_sensor = self.metrics.sensor(sensor_name)
+        self.add_metric('record-error-rate', Rate(),
+                        sensor_name=sensor_name,
+                        description='The average per-second number of record sends that resulted in errors')
+
+        sensor_name = 'record-size-max'
+        self.max_record_size_sensor = self.metrics.sensor(sensor_name)
+        self.add_metric('record-size-max', Max(),
+                        sensor_name=sensor_name,
+                        description='The maximum record size across all batches')
+        self.add_metric('record-size-avg', Avg(),
+                        sensor_name=sensor_name,
+                        description='The average maximum record size per batch')
+
+        self.add_metric('requests-in-flight',
+                        AnonMeasurable(lambda *_: self._client.in_flight_request_count()),
+                        description='The current number of in-flight requests awaiting a response.')
+
+        self.add_metric('metadata-age',
+                        AnonMeasurable(lambda _, now: (now - self._metadata._last_successful_refresh_ms) / 1000),
+                        description='The age in seconds of the current producer metadata being used.')
+
+    def add_metric(self, metric_name, measurable, group_name='producer-metrics',
+                   description=None, tags=None,
+                   sensor_name=None):
+        m = self.metrics
+        metric = m.metric_name(metric_name, group_name, description, tags)
+        if sensor_name:
+            sensor = m.sensor(sensor_name)
+            sensor.add(metric, measurable)
+        else:
+            m.add_metric(metric, measurable)
+
+    def maybe_register_topic_metrics(self, topic):
+
+        def sensor_name(name):
+            return 'topic.{0}.{1}'.format(topic, name)
+
+        # if one sensor of the metrics has been registered for the topic,
+        # then all other sensors should have been registered; and vice versa
+        if not self.metrics.get_sensor(sensor_name('records-per-batch')):
+
+            self.add_metric('record-send-rate', Rate(),
+                            sensor_name=sensor_name('records-per-batch'),
+                            group_name='producer-topic-metrics.' + topic,
+                            description= 'Records sent per second for topic ' + topic)
+
+            self.add_metric('byte-rate', Rate(),
+                            sensor_name=sensor_name('bytes'),
+                            group_name='producer-topic-metrics.' + topic,
+                            description='Bytes per second for topic ' + topic)
+
+            self.add_metric('compression-rate', Avg(),
+                            sensor_name=sensor_name('compression-rate'),
+                            group_name='producer-topic-metrics.' + topic,
+                            description='Average Compression ratio for topic ' + topic)
+
+            self.add_metric('record-retry-rate', Rate(),
+                            sensor_name=sensor_name('record-retries'),
+                            group_name='producer-topic-metrics.' + topic,
+                            description='Record retries per second for topic ' + topic)
+
+            self.add_metric('record-error-rate', Rate(),
+                            sensor_name=sensor_name('record-errors'),
+                            group_name='producer-topic-metrics.' + topic,
+                            description='Record errors per second for topic ' + topic)
+
+    def update_produce_request_metrics(self, batches_map):
+        for node_batch in batches_map.values():
+            records = 0
+            total_bytes = 0
+            for batch in node_batch:
+                # register all per-topic metrics at once
+                topic = batch.topic_partition.topic
+                self.maybe_register_topic_metrics(topic)
+
+                # per-topic record send rate
+                topic_records_count = self.metrics.get_sensor(
+                    'topic.' + topic + '.records-per-batch')
+                topic_records_count.record(batch.record_count)
+
+                # per-topic bytes send rate
+                topic_byte_rate = self.metrics.get_sensor(
+                    'topic.' + topic + '.bytes')
+                topic_byte_rate.record(batch.records.size_in_bytes())
+
+                # per-topic compression rate
+                topic_compression_rate = self.metrics.get_sensor(
+                    'topic.' + topic + '.compression-rate')
+                topic_compression_rate.record(batch.records.compression_rate())
+
+                # global metrics
+                self.batch_size_sensor.record(batch.records.size_in_bytes())
+                if batch.drained:
+                    self.queue_time_sensor.record(batch.drained - batch.created)
+                self.compression_rate_sensor.record(batch.records.compression_rate())
+                self.max_record_size_sensor.record(batch.max_record_size)
+                records += batch.record_count
+                total_bytes += batch.records.size_in_bytes()
+
+            self.records_per_request_sensor.record(records)
+            self.byte_rate_sensor.record(total_bytes)
+
+    def record_retries(self, topic, count):
+        self.retry_sensor.record(count)
+        sensor = self.metrics.get_sensor('topic.' + topic + '.record-retries')
+        if sensor:
+            sensor.record(count)
+
+    def record_errors(self, topic, count):
+        self.error_sensor.record(count)
+        sensor = self.metrics.get_sensor('topic.' + topic + '.record-errors')
+        if sensor:
+            sensor.record(count)
+
+    def record_latency(self, latency, node=None):
+        self.request_time_sensor.record(latency)
+        if node:
+            sensor = self.metrics.get_sensor('node-' + node + '.latency')
+            if sensor:
+                sensor.record(latency)
+
+    def record_throttle_time(self, throttle_time_ms, node=None):
+        self.produce_throttle_time_sensor.record(throttle_time_ms)
diff --git a/kafka/protocol/metadata.py b/kafka/protocol/metadata.py
index 8063dda..2711abb 100644
--- a/kafka/protocol/metadata.py
+++ b/kafka/protocol/metadata.py
@@ -1,5 +1,5 @@
 from .struct import Struct
-from .types import Array, Int16, Int32, Schema, String
+from .types import Array, Boolean, Int16, Int32, Schema, String
 
 
 class MetadataResponse_v0(Struct):
@@ -22,14 +22,46 @@ class MetadataResponse_v0(Struct):
     )
 
 
+class MetadataResponse_v1(Struct):
+    API_KEY = 3
+    API_VERSION = 1
+    SCHEMA = Schema(
+        ('brokers', Array(
+            ('node_id', Int32),
+            ('host', String('utf-8')),
+            ('port', Int32),
+            ('rack', String('utf-8')))),
+        ('controller_id', Int32),
+        ('topics', Array(
+            ('error_code', Int16),
+            ('topic', String('utf-8')),
+            ('is_internal', Boolean),
+            ('partitions', Array(
+                ('error_code', Int16),
+                ('partition', Int32),
+                ('leader', Int32),
+                ('replicas', Array(Int32)),
+                ('isr', Array(Int32))))))
+    )
+
+
 class MetadataRequest_v0(Struct):
     API_KEY = 3
     API_VERSION = 0
     RESPONSE_TYPE = MetadataResponse_v0
     SCHEMA = Schema(
-        ('topics', Array(String('utf-8')))
+        ('topics', Array(String('utf-8'))) # Empty Array (len 0) for all topics
+    )
+
+
+class MetadataRequest_v1(Struct):
+    API_KEY = 3
+    API_VERSION = 1
+    RESPONSE_TYPE = MetadataResponse_v1
+    SCHEMA = Schema(
+        ('topics', Array(String('utf-8'))) # Null Array (len -1) for all topics
     )
 
 
-MetadataRequest = [MetadataRequest_v0]
-MetadataResponse = [MetadataResponse_v0]
+MetadataRequest = [MetadataRequest_v0, MetadataRequest_v1]
+MetadataResponse = [MetadataResponse_v0, MetadataResponse_v1]
diff --git a/kafka/protocol/types.py b/kafka/protocol/types.py
index 18aaca1..da10326 100644
--- a/kafka/protocol/types.py
+++ b/kafka/protocol/types.py
@@ -99,6 +99,16 @@ class Bytes(AbstractType):
         return value
 
 
+class Boolean(AbstractType):
+    @classmethod
+    def encode(cls, value):
+        return _pack('>?', value)
+
+    @classmethod
+    def decode(cls, data):
+        return _unpack('>?', data.read(1))
+
+
 class Schema(AbstractType):
     def __init__(self, *fields):
         if fields:
@@ -145,6 +155,8 @@ class Array(AbstractType):
             raise ValueError('Array instantiated with no array_of type')
 
     def encode(self, items):
+        if items is None:
+            return Int32.encode(-1)
         return b''.join(
             [Int32.encode(len(items))] +
             [self.array_of.encode(item) for item in items]
@@ -152,7 +164,11 @@ class Array(AbstractType):
 
     def decode(self, data):
         length = Int32.decode(data)
+        if length == -1:
+            return None
         return [self.array_of.decode(data) for _ in range(length)]
 
     def repr(self, list_of_items):
+        if list_of_items is None:
+            return 'NULL'
         return '[' + ', '.join([self.array_of.repr(item) for item in list_of_items]) + ']'
diff --git a/kafka/structs.py b/kafka/structs.py
index 5902930..3188516 100644
--- a/kafka/structs.py
+++ b/kafka/structs.py
@@ -58,7 +58,7 @@ TopicPartition = namedtuple("TopicPartition",
     ["topic", "partition"])
 
 BrokerMetadata = namedtuple("BrokerMetadata",
-    ["nodeId", "host", "port"])
+    ["nodeId", "host", "port", "rack"])
 
 PartitionMetadata = namedtuple("PartitionMetadata",
     ["topic", "partition", "leader", "replicas", "isr", "error"])


Test patch:
diff --git a/test/test_client.py b/test/test_client.py
index 660af61..79ac8be 100644
--- a/test/test_client.py
+++ b/test/test_client.py
@@ -1,6 +1,7 @@
 import socket
 
 from mock import ANY, MagicMock, patch
+from operator import itemgetter
 import six
 from . import unittest
 
@@ -117,9 +118,10 @@ class TestSimpleClient(unittest.TestCase):
         mock_conn(conn)
 
         brokers = [
-            BrokerMetadata(0, 'broker_1', 4567),
-            BrokerMetadata(1, 'broker_2', 5678)
+            BrokerMetadata(0, 'broker_1', 4567, None),
+            BrokerMetadata(1, 'broker_2', 5678, None)
         ]
+        resp0_brokers = list(map(itemgetter(0, 1, 2), brokers))
 
         topics = [
             (NO_ERROR, 'topic_1', [
@@ -137,7 +139,7 @@ class TestSimpleClient(unittest.TestCase):
                 (NO_ERROR, 2, 0, [0, 1], [0, 1])
             ])
         ]
-        protocol.decode_metadata_response.return_value = MetadataResponse[0](brokers, topics)
+        protocol.decode_metadata_response.return_value = MetadataResponse[0](resp0_brokers, topics)
 
         # client loads metadata at init
         client = SimpleClient(hosts=['broker_1:4567'])
@@ -167,9 +169,10 @@ class TestSimpleClient(unittest.TestCase):
         mock_conn(conn)
 
         brokers = [
-            BrokerMetadata(0, 'broker_1', 4567),
-            BrokerMetadata(1, 'broker_2', 5678)
+            BrokerMetadata(0, 'broker_1', 4567, None),
+            BrokerMetadata(1, 'broker_2', 5678, None)
         ]
+        resp0_brokers = list(map(itemgetter(0, 1, 2), brokers))
 
         topics = [
             (NO_LEADER, 'topic_still_creating', []),
@@ -179,7 +182,7 @@ class TestSimpleClient(unittest.TestCase):
                 (NO_LEADER, 1, -1, [], []),
             ]),
         ]
-        protocol.decode_metadata_response.return_value = MetadataResponse[0](brokers, topics)
+        protocol.decode_metadata_response.return_value = MetadataResponse[0](resp0_brokers, topics)
 
         client = SimpleClient(hosts=['broker_1:4567'])
 
@@ -197,9 +200,10 @@ class TestSimpleClient(unittest.TestCase):
         mock_conn(conn)
 
         brokers = [
-            BrokerMetadata(0, 'broker_1', 4567),
-            BrokerMetadata(1, 'broker_2', 5678)
+            BrokerMetadata(0, 'broker_1', 4567, None),
+            BrokerMetadata(1, 'broker_2', 5678, None)
         ]
+        resp0_brokers = list(map(itemgetter(0, 1, 2), brokers))
 
         topics = [
             (NO_LEADER, 'topic_still_creating', []),
@@ -209,7 +213,7 @@ class TestSimpleClient(unittest.TestCase):
                 (NO_LEADER, 1, -1, [], []),
             ]),
         ]
-        decode_metadata_response.return_value = MetadataResponse[0](brokers, topics)
+        decode_metadata_response.return_value = MetadataResponse[0](resp0_brokers, topics)
 
         client = SimpleClient(hosts=['broker_1:4567'])
 
@@ -230,14 +234,15 @@ class TestSimpleClient(unittest.TestCase):
         mock_conn(conn)
 
         brokers = [
-            BrokerMetadata(0, 'broker_1', 4567),
-            BrokerMetadata(1, 'broker_2', 5678)
+            BrokerMetadata(0, 'broker_1', 4567, None),
+            BrokerMetadata(1, 'broker_2', 5678, None)
         ]
+        resp0_brokers = list(map(itemgetter(0, 1, 2), brokers))
 
         topics = [
             (NO_LEADER, 'topic_no_partitions', [])
         ]
-        protocol.decode_metadata_response.return_value = MetadataResponse[0](brokers, topics)
+        protocol.decode_metadata_response.return_value = MetadataResponse[0](resp0_brokers, topics)
 
         client = SimpleClient(hosts=['broker_1:4567'])
 
@@ -249,7 +254,7 @@ class TestSimpleClient(unittest.TestCase):
                 (NO_ERROR, 0, 0, [0, 1], [0, 1])
             ])
         ]
-        protocol.decode_metadata_response.return_value = MetadataResponse[0](brokers, topics)
+        protocol.decode_metadata_response.return_value = MetadataResponse[0](resp0_brokers, topics)
 
         # calling _get_leader_for_partition (from any broker aware request)
         # will try loading metadata again for the same topic
@@ -267,15 +272,16 @@ class TestSimpleClient(unittest.TestCase):
         mock_conn(conn)
 
         brokers = [
-            BrokerMetadata(0, 'broker_1', 4567),
-            BrokerMetadata(1, 'broker_2', 5678)
+            BrokerMetadata(0, 'broker_1', 4567, None),
+            BrokerMetadata(1, 'broker_2', 5678, None)
         ]
+        resp0_brokers = list(map(itemgetter(0, 1, 2), brokers))
 
         topics = [
             (NO_LEADER, 'topic_no_partitions', []),
             (UNKNOWN_TOPIC_OR_PARTITION, 'topic_unknown', []),
         ]
-        protocol.decode_metadata_response.return_value = MetadataResponse[0](brokers, topics)
+        protocol.decode_metadata_response.return_value = MetadataResponse[0](resp0_brokers, topics)
 
         client = SimpleClient(hosts=['broker_1:4567'])
 
@@ -294,9 +300,10 @@ class TestSimpleClient(unittest.TestCase):
         mock_conn(conn)
 
         brokers = [
-            BrokerMetadata(0, 'broker_1', 4567),
-            BrokerMetadata(1, 'broker_2', 5678)
+            BrokerMetadata(0, 'broker_1', 4567, None),
+            BrokerMetadata(1, 'broker_2', 5678, None)
         ]
+        resp0_brokers = list(map(itemgetter(0, 1, 2), brokers))
 
         topics = [
             (NO_ERROR, 'topic_noleader', [
@@ -304,7 +311,7 @@ class TestSimpleClient(unittest.TestCase):
                 (NO_LEADER, 1, -1, [], []),
             ]),
         ]
-        protocol.decode_metadata_response.return_value = MetadataResponse[0](brokers, topics)
+        protocol.decode_metadata_response.return_value = MetadataResponse[0](resp0_brokers, topics)
 
         client = SimpleClient(hosts=['broker_1:4567'])
         self.assertDictEqual(
@@ -330,7 +337,7 @@ class TestSimpleClient(unittest.TestCase):
                 (NO_ERROR, 1, 1, [1, 0], [1, 0])
             ]),
         ]
-        protocol.decode_metadata_response.return_value = MetadataResponse[0](brokers, topics)
+        protocol.decode_metadata_response.return_value = MetadataResponse[0](resp0_brokers, topics)
         self.assertEqual(brokers[0], client._get_leader_for_partition('topic_noleader', 0))
         self.assertEqual(brokers[1], client._get_leader_for_partition('topic_noleader', 1))
 
@@ -340,9 +347,10 @@ class TestSimpleClient(unittest.TestCase):
         mock_conn(conn)
 
         brokers = [
-            BrokerMetadata(0, 'broker_1', 4567),
-            BrokerMetadata(1, 'broker_2', 5678)
+            BrokerMetadata(0, 'broker_1', 4567, None),
+            BrokerMetadata(1, 'broker_2', 5678, None)
         ]
+        resp0_brokers = list(map(itemgetter(0, 1, 2), brokers))
 
         topics = [
             (NO_ERROR, 'topic_noleader', [
@@ -350,7 +358,7 @@ class TestSimpleClient(unittest.TestCase):
                 (NO_LEADER, 1, -1, [], []),
             ]),
         ]
-        protocol.decode_metadata_response.return_value = MetadataResponse[0](brokers, topics)
+        protocol.decode_metadata_response.return_value = MetadataResponse[0](resp0_brokers, topics)
 
         client = SimpleClient(hosts=['broker_1:4567'])
 
@@ -368,14 +376,15 @@ class TestSimpleClient(unittest.TestCase):
         mock_conn(conn)
 
         brokers = [
-            BrokerMetadata(0, 'broker_1', 4567),
-            BrokerMetadata(1, 'broker_2', 5678)
+            BrokerMetadata(0, 'broker_1', 4567, None),
+            BrokerMetadata(1, 'broker_2', 5678, None)
         ]
+        resp0_brokers = list(map(itemgetter(0, 1, 2), brokers))
 
         topics = [
             (UNKNOWN_TOPIC_OR_PARTITION, 'topic_doesnt_exist', []),
         ]
-        protocol.decode_metadata_response.return_value = MetadataResponse[0](brokers, topics)
+        protocol.decode_metadata_response.return_value = MetadataResponse[0](resp0_brokers, topics)
 
         client = SimpleClient(hosts=['broker_1:4567'])
 
diff --git a/test/test_client_async.py b/test/test_client_async.py
index dfe11ea..aa91704 100644
--- a/test/test_client_async.py
+++ b/test/test_client_async.py
@@ -53,8 +53,8 @@ def test_bootstrap_success(conn):
     conn.connect.assert_called_with()
     conn.send.assert_called_once_with(MetadataRequest[0]([]))
     assert cli._bootstrap_fails == 0
-    assert cli.cluster.brokers() == set([BrokerMetadata(0, 'foo', 12),
-                                         BrokerMetadata(1, 'bar', 34)])
+    assert cli.cluster.brokers() == set([BrokerMetadata(0, 'foo', 12, None),
+                                         BrokerMetadata(1, 'bar', 34, None)])
 
 def test_bootstrap_failure(conn):
     conn.state = ConnectionStates.DISCONNECTED
diff --git a/test/test_sender.py b/test/test_sender.py
index 44105e2..cf911e1 100644
--- a/test/test_sender.py
+++ b/test/test_sender.py
@@ -7,12 +7,13 @@ import pytest
 
 from kafka.client_async import KafkaClient
 from kafka.cluster import ClusterMetadata
-from kafka.producer.buffer import MessageSetBuffer
-from kafka.producer.sender import Sender
-from kafka.producer.record_accumulator import RecordAccumulator, RecordBatch
 import kafka.errors as Errors
 from kafka.future import Future
+from kafka.metrics import Metrics
+from kafka.producer.buffer import MessageSetBuffer
 from kafka.protocol.produce import ProduceRequest
+from kafka.producer.record_accumulator import RecordAccumulator, RecordBatch
+from kafka.producer.sender import Sender
 from kafka.structs import TopicPartition, OffsetAndMetadata
 
 
@@ -29,8 +30,13 @@ def accumulator():
 
 
 @pytest.fixture
-def sender(client, accumulator):
-    return Sender(client, client.cluster, accumulator)
+def metrics():
+    return Metrics()
+
+
+@pytest.fixture
+def sender(client, accumulator, metrics):
+    return Sender(client, client.cluster, accumulator, metrics)
 
 
 @pytest.mark.parametrize(("api_version", "produce_version"), [