Remove old/unused errors; reorder; KafkaTimeout -> retriable

Author: dpkp

kafka/errors.py

@@ -16,21 +16,44 @@ def __str__(self):
                                super(KafkaError, self).__str__())
 
 
+class Cancelled(KafkaError):
+    retriable = True
+
+
+class CommitFailedError(KafkaError):
+    def __init__(self, *args, **kwargs):
+        super(CommitFailedError, self).__init__(
+            """Commit cannot be completed since the group has already
+            rebalanced and assigned the partitions to another member.
+            This means that the time between subsequent calls to poll()
+            was longer than the configured max_poll_interval_ms, which
+            typically implies that the poll loop is spending too much
+            time message processing. You can address this either by
+            increasing the rebalance timeout with max_poll_interval_ms,
+            or by reducing the maximum size of batches returned in poll()
+            with max_poll_records.
+            """, *args, **kwargs)
+
+
+class IllegalArgumentError(KafkaError):
+    pass
+
+
 class IllegalStateError(KafkaError):
     pass
 
 
-class IllegalArgumentError(KafkaError):
+class IncompatibleBrokerVersion(KafkaError):
     pass
 
 
-class NoBrokersAvailable(KafkaError):
-    retriable = True
-    invalid_metadata = True
+class KafkaConfigurationError(KafkaError):
+    pass
 
 
-class NodeNotReadyError(KafkaError):
+class KafkaConnectionError(KafkaError):
     retriable = True
+    invalid_metadata = True
 
 
 class KafkaProtocolError(KafkaError):
@@ -41,47 +64,41 @@ class CorrelationIdError(KafkaProtocolError):
     retriable = True
 
 
-class Cancelled(KafkaError):
+class KafkaTimeoutError(KafkaError):
     retriable = True
 
 
-class TooManyInFlightRequests(KafkaError):
+class MetadataEmptyBrokerList(KafkaError):
     retriable = True
 
 
-class StaleMetadata(KafkaError):
+class NoBrokersAvailable(KafkaError):
     retriable = True
     invalid_metadata = True
 
 
-class MetadataEmptyBrokerList(KafkaError):
+class NodeNotReadyError(KafkaError):
     retriable = True
 
 
-class UnrecognizedBrokerVersion(KafkaError):
+class QuotaViolationError(KafkaError):
     pass
 
 
-class IncompatibleBrokerVersion(KafkaError):
-    pass
+class StaleMetadata(KafkaError):
+    retriable = True
+    invalid_metadata = True
 
 
-class CommitFailedError(KafkaError):
-    def __init__(self, *args, **kwargs):
-        super(CommitFailedError, self).__init__(
-            """Commit cannot be completed since the group has already
-            rebalanced and assigned the partitions to another member.
-            This means that the time between subsequent calls to poll()
-            was longer than the configured max_poll_interval_ms, which
-            typically implies that the poll loop is spending too much
-            time message processing. You can address this either by
-            increasing the rebalance timeout with max_poll_interval_ms,
-            or by reducing the maximum size of batches returned in poll()
-            with max_poll_records.
-            """, *args, **kwargs)
+class TooManyInFlightRequests(KafkaError):
+    retriable = True
+
 
+class UnrecognizedBrokerVersion(KafkaError):
+    pass
 
-class AuthenticationMethodNotSupported(KafkaError):
+
+class UnsupportedCodecError(KafkaError):
     pass
 
 
@@ -97,6 +114,10 @@ def __str__(self):
             super(BrokerResponseError, self).__str__())
 
 
+class AuthorizationError(BrokerResponseError):
+    pass
+
+
 class NoError(BrokerResponseError):
     errno = 0
     message = 'NO_ERROR'
@@ -332,21 +353,21 @@ class InvalidCommitOffsetSizeError(BrokerResponseError):
                    ' because of oversize metadata.')
 
 
-class TopicAuthorizationFailedError(BrokerResponseError):
+class TopicAuthorizationFailedError(AuthorizationError):
     errno = 29
     message = 'TOPIC_AUTHORIZATION_FAILED'
     description = ('Returned by the broker when the client is not authorized to'
                    ' access the requested topic.')
 
 
-class GroupAuthorizationFailedError(BrokerResponseError):
+class GroupAuthorizationFailedError(AuthorizationError):
     errno = 30
     message = 'GROUP_AUTHORIZATION_FAILED'
     description = ('Returned by the broker when the client is not authorized to'
                    ' access a particular groupId.')
 
 
-class ClusterAuthorizationFailedError(BrokerResponseError):
+class ClusterAuthorizationFailedError(AuthorizationError):
     errno = 31
     message = 'CLUSTER_AUTHORIZATION_FAILED'
     description = ('Returned by the broker when the client is not authorized to'
@@ -493,7 +514,7 @@ class TransactionCoordinatorFencedError(BrokerResponseError):
     retriable = False
 
 
-class TransactionalIdAuthorizationFailedError(BrokerResponseError):
+class TransactionalIdAuthorizationFailedError(AuthorizationError):
     errno = 53
     message = 'TRANSACTIONAL_ID_AUTHORIZATION_FAILED'
     description = 'Transactional Id authorization failed.'
@@ -578,7 +599,7 @@ class DelegationTokenRequestNotAllowedError(BrokerResponseError):
     retriable = False
 
 
-class DelegationTokenAuthorizationFailedError(BrokerResponseError):
+class DelegationTokenAuthorizationFailedError(AuthorizationError):
     errno = 65
     message = 'DELEGATION_TOKEN_AUTHORIZATION_FAILED'
     description = 'Delegation Token authorization failed.'
@@ -1027,47 +1048,6 @@ class VoterNotFoundError(BrokerResponseError):
     retriable = False
 
 
-class KafkaUnavailableError(KafkaError):
-    pass
-
-
-class KafkaTimeoutError(KafkaError):
-    pass
-
-
-class FailedPayloadsError(KafkaError):
-    def __init__(self, payload, *args):
-        super(FailedPayloadsError, self).__init__(*args)
-        self.payload = payload
-
-
-class KafkaConnectionError(KafkaError):
-    retriable = True
-    invalid_metadata = True
-
-
-class ProtocolError(KafkaError):
-    pass
-
-
-class UnsupportedCodecError(KafkaError):
-    pass
-
-
-class KafkaConfigurationError(KafkaError):
-    pass
-
-
-class QuotaViolationError(KafkaError):
-    pass
-
-
-class AsyncProducerQueueFull(KafkaError):
-    def __init__(self, failed_msgs, *args):
-        super(AsyncProducerQueueFull, self).__init__(*args)
-        self.failed_msgs = failed_msgs
-
-
 def _iter_broker_errors():
     for name, obj in inspect.getmembers(sys.modules[__name__]):
         if inspect.isclass(obj) and issubclass(obj, BrokerResponseError) and obj != BrokerResponseError: