Fix lint errors

Author: dpkp

kafka/admin/client.py

@@ -274,7 +274,7 @@ def _refresh_controller_id(self, timeout_ms=30000):
                 self._controller_id = controller_id
                 return
             else:
-                raise Errors.NodeNotAvailableError('controller')
+                raise Errors.NodeNotReadyError('controller')
         else:
             raise UnrecognizedBrokerVersion(
                 "Kafka Admin interface cannot determine the controller using MetadataRequest_v{}."

kafka/producer/sender.py

@@ -315,7 +315,7 @@ def _maybe_send_transactional_request(self):
                     return True
 
             except Exception as e:
-                log.warn("%s: Got an exception when trying to find a node to send a transactional request to. Going to back off and retry", str(self), e)
+                log.warn("%s: Got an exception when trying to find a node to send a transactional request to. Going to back off and retry: %s", str(self), e)
                 if next_request_handler.needs_coordinator():
                     self._transaction_manager.lookup_coordinator_for_request(next_request_handler)
                     break

kafka/producer/transaction_manager.py

@@ -260,7 +260,7 @@ def transition_to_abortable_error(self, exc):
         with self._lock:
             if self._current_state == TransactionState.ABORTING_TRANSACTION:
                 log.debug("Skipping transition to abortable error state since the transaction is already being "
-                          " aborted. Underlying exception: ", exc)
+                          " aborted. Underlying exception: %s", exc)
                 return
             self._transition_to(TransactionState.ABORTABLE_ERROR, error=exc)
 
@@ -687,7 +687,7 @@ def handle_response(self, response):
             if error is Errors.NoError:
                 continue
             elif error in (Errors.CoordinatorNotAvailableError, Errors.NotCoordinatorError):
-                self.transaction_manager._lookup_coordinator('transaction', self.transactiona_id)
+                self.transaction_manager._lookup_coordinator('transaction', self.transactional_id)
                 self.reenqueue()
                 return
             elif error is Errors.ConcurrentTransactionsError:
@@ -726,7 +726,7 @@ def handle_response(self, response):
         self.transaction_manager._pending_partitions_in_transaction -= partitions
 
         if unauthorized_topics:
-            self.abortable_error(Errors.TopicAuthorizationError(unauthorized_topics))
+            self.abortable_error(Errors.TopicAuthorizationFailedError(unauthorized_topics))
         elif has_partition_errors:
             self.abortable_error(Errors.KafkaError("Could not add partitions to transaction due to errors: %s" % (results)))
         else:
@@ -795,7 +795,7 @@ def handle_response(self, response):
         elif error is Errors.TransactionalIdAuthorizationFailedError:
             self.fatal_error(error())
         elif error is Errors.GroupAuthorizationFailedError:
-            self.abortable_error(Errors.GroupAuthorizationError(self._coord_key))
+            self.abortable_error(error(self._coord_key))
         else:
             self.fatal_error(Errors.KafkaError(
                 "Could not find a coordinator with type %s with key %s due to"
@@ -888,7 +888,7 @@ def handle_response(self, response):
         elif error is Errors.TransactionalIdAuthorizationFailedError:
             self.fatal_error(error())
         elif error is Errors.GroupAuthorizationFailedError:
-            self.abortable_error(Errors.GroupAuthorizationError(self.consumer_group_id))
+            self.abortable_error(error(self.consumer_group_id))
         else:
             self.fatal_error(Errors.KafkaError("Unexpected error in AddOffsetsToTxnResponse: %s" % (error())))
 
@@ -955,7 +955,7 @@ def handle_response(self, response):
             elif error is Errors.UnknownTopicOrPartitionError:
                 retriable_failure = True
             elif error is Errors.GroupAuthorizationFailedError:
-                self.abortable_error(Errors.GroupAuthorizationError(self.consumer_group_id))
+                self.abortable_error(error(self.consumer_group_id))
                 return
             elif error in (Errors.TransactionalIdAuthorizationFailedError,
                            Errors.InvalidProducerEpochError,

kafka/record/default_records.py

@@ -117,6 +117,8 @@ def _assert_has_codec(self, compression_type):
             checker, name = codecs.has_lz4, "lz4"
         elif compression_type == self.CODEC_ZSTD:
             checker, name = codecs.has_zstd, "zstd"
+        else:
+            raise UnsupportedCodecError("Unrecognized compression type: %s" % (compression_type,))
         if not checker():
             raise UnsupportedCodecError(
                 "Libraries for {} compression codec not found".format(name))