diff --git a/kafka/consumer/subscription_state.py b/kafka/consumer/subscription_state.py
index b30922b3e..0cfcfd2d4 100644
--- a/kafka/consumer/subscription_state.py
+++ b/kafka/consumer/subscription_state.py
@@ -421,6 +421,7 @@ def is_fetchable(self):
         return not self.paused and self.has_valid_position
 
 
+@six.add_metaclass(abc.ABCMeta)
 class ConsumerRebalanceListener(object):
     """
     A callback interface that the user can implement to trigger custom actions
@@ -462,8 +463,6 @@ class ConsumerRebalanceListener(object):
     taking over that partition has their on_partitions_assigned() callback
     called to load the state.
     """
-    __metaclass__ = abc.ABCMeta
-
     @abc.abstractmethod
     def on_partitions_revoked(self, revoked):
         """
diff --git a/kafka/metrics/compound_stat.py b/kafka/metrics/compound_stat.py
index ac92480dc..f8b149c54 100644
--- a/kafka/metrics/compound_stat.py
+++ b/kafka/metrics/compound_stat.py
@@ -3,16 +3,16 @@
 import abc
 
 from kafka.metrics.stat import AbstractStat
+from kafka.vendor.six import add_metaclass
 
 
+@add_metaclass(abc.ABCMeta)
 class AbstractCompoundStat(AbstractStat):
     """
     A compound stat is a stat where a single measurement and associated
     data structure feeds many metrics. This is the example for a
     histogram which has many associated percentiles.
     """
-    __metaclass__ = abc.ABCMeta
-
     def stats(self):
         """
         Return list of NamedMeasurable
diff --git a/kafka/metrics/measurable_stat.py b/kafka/metrics/measurable_stat.py
index 4487adf6e..08222b144 100644
--- a/kafka/metrics/measurable_stat.py
+++ b/kafka/metrics/measurable_stat.py
@@ -4,8 +4,10 @@
 
 from kafka.metrics.measurable import AbstractMeasurable
 from kafka.metrics.stat import AbstractStat
+from kafka.vendor.six import add_metaclass
 
 
+@add_metaclass(abc.ABCMeta)
 class AbstractMeasurableStat(AbstractStat, AbstractMeasurable):
     """
     An AbstractMeasurableStat is an AbstractStat that is also
@@ -13,4 +15,3 @@ class AbstractMeasurableStat(AbstractStat, AbstractMeasurable):
     This is the interface used for most of the simple statistics such
     as Avg, Max, Count, etc.
     """
-    __metaclass__ = abc.ABCMeta
diff --git a/kafka/metrics/metrics_reporter.py b/kafka/metrics/metrics_reporter.py
index d8bd12b3b..8df2e9ea6 100644
--- a/kafka/metrics/metrics_reporter.py
+++ b/kafka/metrics/metrics_reporter.py
@@ -2,14 +2,15 @@
 
 import abc
 
+from kafka.vendor.six import add_metaclass
 
+
+@add_metaclass(abc.ABCMeta)
 class AbstractMetricsReporter(object):
     """
     An abstract class to allow things to listen as new metrics
     are created so they can be reported.
     """
-    __metaclass__ = abc.ABCMeta
-
     @abc.abstractmethod
     def init(self, metrics):
         """
diff --git a/kafka/metrics/stat.py b/kafka/metrics/stat.py
index 9fd2f01ec..8825d2783 100644
--- a/kafka/metrics/stat.py
+++ b/kafka/metrics/stat.py
@@ -2,14 +2,15 @@
 
 import abc
 
+from kafka.vendor.six import add_metaclass
 
+
+@add_metaclass(abc.ABCMeta)
 class AbstractStat(object):
     """
     An AbstractStat is a quantity such as average, max, etc that is computed
     off the stream of updates to a sensor
     """
-    __metaclass__ = abc.ABCMeta
-
     @abc.abstractmethod
     def record(self, config, value, time_ms):
         """
diff --git a/kafka/metrics/stats/sampled_stat.py b/kafka/metrics/stats/sampled_stat.py
index c41b14bbc..146687916 100644
--- a/kafka/metrics/stats/sampled_stat.py
+++ b/kafka/metrics/stats/sampled_stat.py
@@ -3,8 +3,10 @@
 import abc
 
 from kafka.metrics.measurable_stat import AbstractMeasurableStat
+from kafka.vendor.six import add_metaclass
 
 
+@add_metaclass(abc.ABCMeta)
 class AbstractSampledStat(AbstractMeasurableStat):
     """
     An AbstractSampledStat records a single scalar value measured over
@@ -20,8 +22,6 @@ class AbstractSampledStat(AbstractMeasurableStat):
     Subclasses of this class define different statistics measured
     using this basic pattern.
     """
-    __metaclass__ = abc.ABCMeta
-
     def __init__(self, initial_value):
         self._initial_value = initial_value
         self._samples = []
diff --git a/kafka/protocol/abstract.py b/kafka/protocol/abstract.py
index 2de65c4bb..7ce5fc18f 100644
--- a/kafka/protocol/abstract.py
+++ b/kafka/protocol/abstract.py
@@ -2,10 +2,11 @@
 
 import abc
 
+from kafka.vendor.six import add_metaclass
 
-class AbstractType(object):
-    __metaclass__ = abc.ABCMeta
 
+@add_metaclass(abc.ABCMeta)
+class AbstractType(object):
     @abc.abstractmethod
     def encode(cls, value): # pylint: disable=no-self-argument
         pass
diff --git a/kafka/protocol/api.py b/kafka/protocol/api.py
index f12cb972b..714da4d1d 100644
--- a/kafka/protocol/api.py
+++ b/kafka/protocol/api.py
@@ -5,6 +5,8 @@
 from kafka.protocol.struct import Struct
 from kafka.protocol.types import Int16, Int32, String, Schema, Array, TaggedFields
 
+from kafka.vendor.six import add_metaclass
+
 
 class RequestHeader(Struct):
     SCHEMA = Schema(
@@ -49,9 +51,8 @@ class ResponseHeaderV2(Struct):
     )
 
 
+@add_metaclass(abc.ABCMeta)
 class Request(Struct):
-    __metaclass__ = abc.ABCMeta
-
     FLEXIBLE_VERSION = False
 
     @abc.abstractproperty
@@ -92,8 +93,8 @@ def parse_response_header(self, read_buffer):
         return ResponseHeader.decode(read_buffer)
 
 
+@add_metaclass(abc.ABCMeta)
 class Response(Struct):
-    __metaclass__ = abc.ABCMeta
 
     @abc.abstractproperty
     def API_KEY(self):
diff --git a/kafka/record/abc.py b/kafka/record/abc.py
index 8509e23e5..73f91a039 100644
--- a/kafka/record/abc.py
+++ b/kafka/record/abc.py
@@ -1,9 +1,12 @@
 from __future__ import absolute_import
+
 import abc
 
+from kafka.vendor.six import add_metaclass
+
 
+@add_metaclass(abc.ABCMeta)
 class ABCRecord(object):
-    __metaclass__ = abc.ABCMeta
     __slots__ = ()
 
     @abc.abstractproperty
@@ -44,8 +47,8 @@ def headers(self):
         """
 
 
+@add_metaclass(abc.ABCMeta)
 class ABCRecordBatchBuilder(object):
-    __metaclass__ = abc.ABCMeta
     __slots__ = ()
 
     @abc.abstractmethod
@@ -84,11 +87,11 @@ def build(self):
         """
 
 
+@add_metaclass(abc.ABCMeta)
 class ABCRecordBatch(object):
     """ For v2 encapsulates a RecordBatch, for v0/v1 a single (maybe
         compressed) message.
     """
-    __metaclass__ = abc.ABCMeta
     __slots__ = ()
 
     @abc.abstractmethod
@@ -98,8 +101,8 @@ def __iter__(self):
         """
 
 
+@add_metaclass(abc.ABCMeta)
 class ABCRecords(object):
-    __metaclass__ = abc.ABCMeta
     __slots__ = ()
 
     @abc.abstractmethod
diff --git a/kafka/sasl/abc.py b/kafka/sasl/abc.py
index 8977c7c23..0577888a9 100644
--- a/kafka/sasl/abc.py
+++ b/kafka/sasl/abc.py
@@ -2,10 +2,11 @@
 
 import abc
 
+from kafka.vendor.six import add_metaclass
 
-class SaslMechanism(object):
-    __metaclass__ = abc.ABCMeta
 
+@add_metaclass(abc.ABCMeta)
+class SaslMechanism(object):
     @abc.abstractmethod
     def __init__(self, **config):
         pass