python3 updates: drop class object, super() (#2837)

Author: dpkp

kafka/admin/acl_resource.py

@@ -160,7 +160,7 @@ def __init__(
             permission_type,
             resource_pattern
     ):
-        super(ACL, self).__init__(principal, host, operation, permission_type, resource_pattern)
+        super().__init__(principal, host, operation, permission_type, resource_pattern)
         self.validate()
 
     def validate(self):
@@ -231,7 +231,7 @@ def __init__(
             resource_name,
             pattern_type=ACLResourcePatternType.LITERAL
     ):
-        super(ResourcePattern, self).__init__(resource_type, resource_name, pattern_type)
+        super().__init__(resource_type, resource_name, pattern_type)
         self.validate()
 
     def validate(self):

kafka/benchmarks/consumer_performance.py

@@ -71,7 +71,7 @@ def run(args):
 
 class StatsReporter(threading.Thread):
     def __init__(self, interval, consumer, event=None, raw_metrics=False):
-        super(StatsReporter, self).__init__()
+        super().__init__()
         self.interval = interval
         self.consumer = consumer
         self.event = event

kafka/benchmarks/load_example.py

@@ -11,7 +11,7 @@
 class Producer(threading.Thread):
 
     def __init__(self, bootstrap_servers, topic, stop_event, msg_size):
-        super(Producer, self).__init__()
+        super().__init__()
         self.bootstrap_servers = bootstrap_servers
         self.topic = topic
         self.stop_event = stop_event
@@ -30,7 +30,7 @@ def run(self):
 
 class Consumer(threading.Thread):
     def __init__(self, bootstrap_servers, topic, stop_event, msg_size):
-        super(Consumer, self).__init__()
+        super().__init__()
         self.bootstrap_servers = bootstrap_servers
         self.topic = topic
         self.stop_event = stop_event

kafka/benchmarks/producer_performance.py

@@ -82,7 +82,7 @@ def _benchmark():
 
 class StatsReporter(threading.Thread):
     def __init__(self, interval, producer, event=None, raw_metrics=False):
-        super(StatsReporter, self).__init__()
+        super().__init__()
         self.interval = interval
         self.producer = producer
         self.event = event

kafka/coordinator/base.py

@@ -1089,7 +1089,7 @@ def __init__(self, heartbeat, metrics, prefix, tags=None):
 
 class HeartbeatThread(threading.Thread):
     def __init__(self, coordinator):
-        super(HeartbeatThread, self).__init__()
+        super().__init__()
         self.name = coordinator.group_id + '-heartbeat'
         self.coordinator = coordinator
         self.enabled = False

kafka/coordinator/consumer.py

@@ -86,7 +86,7 @@ def __init__(self, client, subscription, **configs):
                 True the only way to receive records from an internal topic is
                 subscribing to it. Requires 0.10+. Default: True
         """
-        super(ConsumerCoordinator, self).__init__(client, **configs)
+        super().__init__(client, **configs)
 
         self.config = copy.copy(self.DEFAULT_CONFIG)
         for key in self.config:
@@ -149,7 +149,7 @@ def __del__(self):
                 self._cluster.remove_listener(WeakMethod(self._handle_metadata_update))
             except TypeError:
                 pass
-        super(ConsumerCoordinator, self).__del__()
+        super().__del__()
 
     def protocol_type(self):
         return ConsumerProtocolType
@@ -403,7 +403,7 @@ def need_rejoin(self):
             and self._joined_subscription != self._subscription.subscription):
             return True
 
-        return super(ConsumerCoordinator, self).need_rejoin()
+        return super().need_rejoin()
 
     def refresh_committed_offsets_if_needed(self, timeout_ms=None):
         """Fetch committed offsets for assigned partitions."""
@@ -477,7 +477,7 @@ def close(self, autocommit=True, timeout_ms=None):
             if autocommit:
                 self._maybe_auto_commit_offsets_sync(timeout_ms=timeout_ms)
         finally:
-            super(ConsumerCoordinator, self).close(timeout_ms=timeout_ms)
+            super().close(timeout_ms=timeout_ms)
 
     def _invoke_completed_offset_commit_callbacks(self):
         if self._async_commit_fenced:

kafka/errors.py

@@ -11,7 +11,7 @@ def __str__(self):
         if not self.args:
             return self.__class__.__name__
         return '{0}: {1}'.format(self.__class__.__name__,
-                               super(KafkaError, self).__str__())
+                               super().__str__())
 
     def __eq__(self, other):
         return self.__class__ == other.__class__ and self.args == other.args
@@ -26,7 +26,7 @@ def __init__(self, *args):
         if not args:
             args = ("Commit cannot be completed since the group has already"
                     " rebalanced and assigned the partitions to another member.",)
-        super(CommitFailedError, self).__init__(*args)
+        super().__init__(*args)
 
 
 class IllegalArgumentError(KafkaError):
@@ -113,7 +113,7 @@ def __str__(self):
         """Add errno to standard KafkaError str"""
         return '[Error {0}] {1}'.format(
             self.errno,
-            super(BrokerResponseError, self).__str__())
+            super().__str__())
 
 
 class AuthorizationError(BrokerResponseError):

kafka/producer/future.py

@@ -7,17 +7,17 @@
 
 class FutureProduceResult(Future):
     def __init__(self, topic_partition):
-        super(FutureProduceResult, self).__init__()
+        super().__init__()
         self.topic_partition = topic_partition
         self._latch = threading.Event()
 
     def success(self, value):
-        ret = super(FutureProduceResult, self).success(value)
+        ret = super().success(value)
         self._latch.set()
         return ret
 
     def failure(self, error):
-        ret = super(FutureProduceResult, self).failure(error)
+        ret = super().failure(error)
         self._latch.set()
         return ret
 
@@ -28,7 +28,7 @@ def wait(self, timeout=None):
 
 class FutureRecordMetadata(Future):
     def __init__(self, produce_future, batch_index, timestamp_ms, checksum, serialized_key_size, serialized_value_size, serialized_header_size):
-        super(FutureRecordMetadata, self).__init__()
+        super().__init__()
         self._produce_future = produce_future
         # packing args as a tuple is a minor speed optimization
         self.args = (batch_index, timestamp_ms, checksum, serialized_key_size, serialized_value_size, serialized_header_size)

kafka/producer/sender.py

@@ -42,7 +42,7 @@ class Sender(threading.Thread):
     }
 
     def __init__(self, client, metadata, accumulator, **configs):
-        super(Sender, self).__init__()
+        super().__init__()
         self.config = copy.copy(self.DEFAULT_CONFIG)
         for key in self.config:
             if key in configs:

kafka/producer/transaction_manager.py

@@ -524,7 +524,7 @@ def exception(self):
         return self._error
 
 
-class TxnRequestHandler(object, metaclass=abc.ABCMeta):
+class TxnRequestHandler(metaclass=abc.ABCMeta):
     def __init__(self, transaction_manager, result=None):
         self.transaction_manager = transaction_manager
         self.retry_backoff_ms = transaction_manager.retry_backoff_ms
@@ -612,7 +612,7 @@ def priority(self):
 
 class InitProducerIdHandler(TxnRequestHandler):
     def __init__(self, transaction_manager, transaction_timeout_ms):
-        super(InitProducerIdHandler, self).__init__(transaction_manager)
+        super().__init__(transaction_manager)
 
         if transaction_manager._api_version >= (2, 0):
             version = 1
@@ -645,7 +645,7 @@ def handle_response(self, response):
 
 class AddPartitionsToTxnHandler(TxnRequestHandler):
     def __init__(self, transaction_manager, topic_partitions):
-        super(AddPartitionsToTxnHandler, self).__init__(transaction_manager)
+        super().__init__(transaction_manager)
 
         if transaction_manager._api_version >= (2, 7):
             version = 2
@@ -740,7 +740,7 @@ def maybe_override_retry_backoff_ms(self):
 
 class FindCoordinatorHandler(TxnRequestHandler):
     def __init__(self, transaction_manager, coord_type, coord_key):
-        super(FindCoordinatorHandler, self).__init__(transaction_manager)
+        super().__init__(transaction_manager)
 
         self._coord_type = coord_type
         self._coord_key = coord_key
@@ -796,7 +796,7 @@ def handle_response(self, response):
 
 class EndTxnHandler(TxnRequestHandler):
     def __init__(self, transaction_manager, committed):
-        super(EndTxnHandler, self).__init__(transaction_manager)
+        super().__init__(transaction_manager)
 
         if self.transaction_manager._api_version >= (2, 7):
             version = 2
@@ -837,7 +837,7 @@ def handle_response(self, response):
 
 class AddOffsetsToTxnHandler(TxnRequestHandler):
     def __init__(self, transaction_manager, consumer_group_id, offsets):
-        super(AddOffsetsToTxnHandler, self).__init__(transaction_manager)
+        super().__init__(transaction_manager)
 
         self.consumer_group_id = consumer_group_id
         self.offsets = offsets
@@ -887,7 +887,7 @@ def handle_response(self, response):
 
 class TxnOffsetCommitHandler(TxnRequestHandler):
     def __init__(self, transaction_manager, consumer_group_id, offsets, result):
-        super(TxnOffsetCommitHandler, self).__init__(transaction_manager, result=result)
+        super().__init__(transaction_manager, result=result)
 
         self.consumer_group_id = consumer_group_id
         self.offsets = offsets