errors.go 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369
  1. package sarama
  2. import (
  3. "errors"
  4. "fmt"
  5. )
  6. // ErrOutOfBrokers is the error returned when the client has run out of brokers to talk to because all of them errored
  7. // or otherwise failed to respond.
  8. var ErrOutOfBrokers = errors.New("kafka: client has run out of available brokers to talk to (Is your cluster reachable?)")
  9. // ErrClosedClient is the error returned when a method is called on a client that has been closed.
  10. var ErrClosedClient = errors.New("kafka: tried to use a client that was closed")
  11. // ErrIncompleteResponse is the error returned when the server returns a syntactically valid response, but it does
  12. // not contain the expected information.
  13. var ErrIncompleteResponse = errors.New("kafka: response did not contain all the expected topic/partition blocks")
  14. // ErrInvalidPartition is the error returned when a partitioner returns an invalid partition index
  15. // (meaning one outside of the range [0...numPartitions-1]).
  16. var ErrInvalidPartition = errors.New("kafka: partitioner returned an invalid partition index")
  17. // ErrAlreadyConnected is the error returned when calling Open() on a Broker that is already connected or connecting.
  18. var ErrAlreadyConnected = errors.New("kafka: broker connection already initiated")
  19. // ErrNotConnected is the error returned when trying to send or call Close() on a Broker that is not connected.
  20. var ErrNotConnected = errors.New("kafka: broker not connected")
  21. // ErrInsufficientData is returned when decoding and the packet is truncated. This can be expected
  22. // when requesting messages, since as an optimization the server is allowed to return a partial message at the end
  23. // of the message set.
  24. var ErrInsufficientData = errors.New("kafka: insufficient data to decode packet, more bytes expected")
  25. // ErrShuttingDown is returned when a producer receives a message during shutdown.
  26. var ErrShuttingDown = errors.New("kafka: message received by producer in process of shutting down")
  27. // ErrMessageTooLarge is returned when the next message to consume is larger than the configured Consumer.Fetch.Max
  28. var ErrMessageTooLarge = errors.New("kafka: message is larger than Consumer.Fetch.Max")
  29. // ErrConsumerOffsetNotAdvanced is returned when a partition consumer didn't advance its offset after parsing
  30. // a RecordBatch.
  31. var ErrConsumerOffsetNotAdvanced = errors.New("kafka: consumer offset was not advanced after a RecordBatch")
  32. // ErrControllerNotAvailable is returned when server didn't give correct controller id. May be kafka server's version
  33. // is lower than 0.10.0.0.
  34. var ErrControllerNotAvailable = errors.New("kafka: controller is not available")
  35. // ErrNoTopicsToUpdateMetadata is returned when Meta.Full is set to false but no specific topics were found to update
  36. // the metadata.
  37. var ErrNoTopicsToUpdateMetadata = errors.New("kafka: no specific topics to update metadata")
  38. // PacketEncodingError is returned from a failure while encoding a Kafka packet. This can happen, for example,
  39. // if you try to encode a string over 2^15 characters in length, since Kafka's encoding rules do not permit that.
  40. type PacketEncodingError struct {
  41. Info string
  42. }
  43. func (err PacketEncodingError) Error() string {
  44. return fmt.Sprintf("kafka: error encoding packet: %s", err.Info)
  45. }
  46. // PacketDecodingError is returned when there was an error (other than truncated data) decoding the Kafka broker's response.
  47. // This can be a bad CRC or length field, or any other invalid value.
  48. type PacketDecodingError struct {
  49. Info string
  50. }
  51. func (err PacketDecodingError) Error() string {
  52. return fmt.Sprintf("kafka: error decoding packet: %s", err.Info)
  53. }
  54. // ConfigurationError is the type of error returned from a constructor (e.g. NewClient, or NewConsumer)
  55. // when the specified configuration is invalid.
  56. type ConfigurationError string
  57. func (err ConfigurationError) Error() string {
  58. return "kafka: invalid configuration (" + string(err) + ")"
  59. }
  60. // KError is the type of error that can be returned directly by the Kafka broker.
  61. // See https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ErrorCodes
  62. type KError int16
  63. // MultiError is used to contain multi error.
  64. type MultiError struct {
  65. Errors *[]error
  66. }
  67. func (mErr MultiError) Error() string {
  68. var errString = ""
  69. for _, err := range *mErr.Errors {
  70. errString += err.Error() + ","
  71. }
  72. return errString
  73. }
  74. // ErrDeleteRecords is the type of error returned when fail to delete the required records
  75. type ErrDeleteRecords struct {
  76. MultiError
  77. }
  78. func (err ErrDeleteRecords) Error() string {
  79. return "kafka server: failed to delete records " + err.MultiError.Error()
  80. }
  81. // Numeric error codes returned by the Kafka server.
  82. const (
  83. ErrNoError KError = 0
  84. ErrUnknown KError = -1
  85. ErrOffsetOutOfRange KError = 1
  86. ErrInvalidMessage KError = 2
  87. ErrUnknownTopicOrPartition KError = 3
  88. ErrInvalidMessageSize KError = 4
  89. ErrLeaderNotAvailable KError = 5
  90. ErrNotLeaderForPartition KError = 6
  91. ErrRequestTimedOut KError = 7
  92. ErrBrokerNotAvailable KError = 8
  93. ErrReplicaNotAvailable KError = 9
  94. ErrMessageSizeTooLarge KError = 10
  95. ErrStaleControllerEpochCode KError = 11
  96. ErrOffsetMetadataTooLarge KError = 12
  97. ErrNetworkException KError = 13
  98. ErrOffsetsLoadInProgress KError = 14
  99. ErrConsumerCoordinatorNotAvailable KError = 15
  100. ErrNotCoordinatorForConsumer KError = 16
  101. ErrInvalidTopic KError = 17
  102. ErrMessageSetSizeTooLarge KError = 18
  103. ErrNotEnoughReplicas KError = 19
  104. ErrNotEnoughReplicasAfterAppend KError = 20
  105. ErrInvalidRequiredAcks KError = 21
  106. ErrIllegalGeneration KError = 22
  107. ErrInconsistentGroupProtocol KError = 23
  108. ErrInvalidGroupId KError = 24
  109. ErrUnknownMemberId KError = 25
  110. ErrInvalidSessionTimeout KError = 26
  111. ErrRebalanceInProgress KError = 27
  112. ErrInvalidCommitOffsetSize KError = 28
  113. ErrTopicAuthorizationFailed KError = 29
  114. ErrGroupAuthorizationFailed KError = 30
  115. ErrClusterAuthorizationFailed KError = 31
  116. ErrInvalidTimestamp KError = 32
  117. ErrUnsupportedSASLMechanism KError = 33
  118. ErrIllegalSASLState KError = 34
  119. ErrUnsupportedVersion KError = 35
  120. ErrTopicAlreadyExists KError = 36
  121. ErrInvalidPartitions KError = 37
  122. ErrInvalidReplicationFactor KError = 38
  123. ErrInvalidReplicaAssignment KError = 39
  124. ErrInvalidConfig KError = 40
  125. ErrNotController KError = 41
  126. ErrInvalidRequest KError = 42
  127. ErrUnsupportedForMessageFormat KError = 43
  128. ErrPolicyViolation KError = 44
  129. ErrOutOfOrderSequenceNumber KError = 45
  130. ErrDuplicateSequenceNumber KError = 46
  131. ErrInvalidProducerEpoch KError = 47
  132. ErrInvalidTxnState KError = 48
  133. ErrInvalidProducerIDMapping KError = 49
  134. ErrInvalidTransactionTimeout KError = 50
  135. ErrConcurrentTransactions KError = 51
  136. ErrTransactionCoordinatorFenced KError = 52
  137. ErrTransactionalIDAuthorizationFailed KError = 53
  138. ErrSecurityDisabled KError = 54
  139. ErrOperationNotAttempted KError = 55
  140. ErrKafkaStorageError KError = 56
  141. ErrLogDirNotFound KError = 57
  142. ErrSASLAuthenticationFailed KError = 58
  143. ErrUnknownProducerID KError = 59
  144. ErrReassignmentInProgress KError = 60
  145. ErrDelegationTokenAuthDisabled KError = 61
  146. ErrDelegationTokenNotFound KError = 62
  147. ErrDelegationTokenOwnerMismatch KError = 63
  148. ErrDelegationTokenRequestNotAllowed KError = 64
  149. ErrDelegationTokenAuthorizationFailed KError = 65
  150. ErrDelegationTokenExpired KError = 66
  151. ErrInvalidPrincipalType KError = 67
  152. ErrNonEmptyGroup KError = 68
  153. ErrGroupIDNotFound KError = 69
  154. ErrFetchSessionIDNotFound KError = 70
  155. ErrInvalidFetchSessionEpoch KError = 71
  156. ErrListenerNotFound KError = 72
  157. ErrTopicDeletionDisabled KError = 73
  158. ErrFencedLeaderEpoch KError = 74
  159. ErrUnknownLeaderEpoch KError = 75
  160. ErrUnsupportedCompressionType KError = 76
  161. ErrStaleBrokerEpoch KError = 77
  162. ErrOffsetNotAvailable KError = 78
  163. ErrMemberIdRequired KError = 79
  164. ErrPreferredLeaderNotAvailable KError = 80
  165. ErrGroupMaxSizeReached KError = 81
  166. ErrFencedInstancedId KError = 82
  167. )
  168. func (err KError) Error() string {
  169. // Error messages stolen/adapted from
  170. // https://kafka.apache.org/protocol#protocol_error_codes
  171. switch err {
  172. case ErrNoError:
  173. return "kafka server: Not an error, why are you printing me?"
  174. case ErrUnknown:
  175. return "kafka server: Unexpected (unknown?) server error."
  176. case ErrOffsetOutOfRange:
  177. return "kafka server: The requested offset is outside the range of offsets maintained by the server for the given topic/partition."
  178. case ErrInvalidMessage:
  179. return "kafka server: Message contents does not match its CRC."
  180. case ErrUnknownTopicOrPartition:
  181. return "kafka server: Request was for a topic or partition that does not exist on this broker."
  182. case ErrInvalidMessageSize:
  183. return "kafka server: The message has a negative size."
  184. case ErrLeaderNotAvailable:
  185. return "kafka server: In the middle of a leadership election, there is currently no leader for this partition and hence it is unavailable for writes."
  186. case ErrNotLeaderForPartition:
  187. return "kafka server: Tried to send a message to a replica that is not the leader for some partition. Your metadata is out of date."
  188. case ErrRequestTimedOut:
  189. return "kafka server: Request exceeded the user-specified time limit in the request."
  190. case ErrBrokerNotAvailable:
  191. return "kafka server: Broker not available. Not a client facing error, we should never receive this!!!"
  192. case ErrReplicaNotAvailable:
  193. return "kafka server: Replica information not available, one or more brokers are down."
  194. case ErrMessageSizeTooLarge:
  195. return "kafka server: Message was too large, server rejected it to avoid allocation error."
  196. case ErrStaleControllerEpochCode:
  197. return "kafka server: StaleControllerEpochCode (internal error code for broker-to-broker communication)."
  198. case ErrOffsetMetadataTooLarge:
  199. return "kafka server: Specified a string larger than the configured maximum for offset metadata."
  200. case ErrNetworkException:
  201. return "kafka server: The server disconnected before a response was received."
  202. case ErrOffsetsLoadInProgress:
  203. return "kafka server: The broker is still loading offsets after a leader change for that offset's topic partition."
  204. case ErrConsumerCoordinatorNotAvailable:
  205. return "kafka server: Offset's topic has not yet been created."
  206. case ErrNotCoordinatorForConsumer:
  207. return "kafka server: Request was for a consumer group that is not coordinated by this broker."
  208. case ErrInvalidTopic:
  209. return "kafka server: The request attempted to perform an operation on an invalid topic."
  210. case ErrMessageSetSizeTooLarge:
  211. return "kafka server: The request included message batch larger than the configured segment size on the server."
  212. case ErrNotEnoughReplicas:
  213. return "kafka server: Messages are rejected since there are fewer in-sync replicas than required."
  214. case ErrNotEnoughReplicasAfterAppend:
  215. return "kafka server: Messages are written to the log, but to fewer in-sync replicas than required."
  216. case ErrInvalidRequiredAcks:
  217. return "kafka server: The number of required acks is invalid (should be either -1, 0, or 1)."
  218. case ErrIllegalGeneration:
  219. return "kafka server: The provided generation id is not the current generation."
  220. case ErrInconsistentGroupProtocol:
  221. return "kafka server: The provider group protocol type is incompatible with the other members."
  222. case ErrInvalidGroupId:
  223. return "kafka server: The provided group id was empty."
  224. case ErrUnknownMemberId:
  225. return "kafka server: The provided member is not known in the current generation."
  226. case ErrInvalidSessionTimeout:
  227. return "kafka server: The provided session timeout is outside the allowed range."
  228. case ErrRebalanceInProgress:
  229. return "kafka server: A rebalance for the group is in progress. Please re-join the group."
  230. case ErrInvalidCommitOffsetSize:
  231. return "kafka server: The provided commit metadata was too large."
  232. case ErrTopicAuthorizationFailed:
  233. return "kafka server: The client is not authorized to access this topic."
  234. case ErrGroupAuthorizationFailed:
  235. return "kafka server: The client is not authorized to access this group."
  236. case ErrClusterAuthorizationFailed:
  237. return "kafka server: The client is not authorized to send this request type."
  238. case ErrInvalidTimestamp:
  239. return "kafka server: The timestamp of the message is out of acceptable range."
  240. case ErrUnsupportedSASLMechanism:
  241. return "kafka server: The broker does not support the requested SASL mechanism."
  242. case ErrIllegalSASLState:
  243. return "kafka server: Request is not valid given the current SASL state."
  244. case ErrUnsupportedVersion:
  245. return "kafka server: The version of API is not supported."
  246. case ErrTopicAlreadyExists:
  247. return "kafka server: Topic with this name already exists."
  248. case ErrInvalidPartitions:
  249. return "kafka server: Number of partitions is invalid."
  250. case ErrInvalidReplicationFactor:
  251. return "kafka server: Replication-factor is invalid."
  252. case ErrInvalidReplicaAssignment:
  253. return "kafka server: Replica assignment is invalid."
  254. case ErrInvalidConfig:
  255. return "kafka server: Configuration is invalid."
  256. case ErrNotController:
  257. return "kafka server: This is not the correct controller for this cluster."
  258. case ErrInvalidRequest:
  259. return "kafka server: This most likely occurs because of a request being malformed by the client library or the message was sent to an incompatible broker. See the broker logs for more details."
  260. case ErrUnsupportedForMessageFormat:
  261. return "kafka server: The requested operation is not supported by the message format version."
  262. case ErrPolicyViolation:
  263. return "kafka server: Request parameters do not satisfy the configured policy."
  264. case ErrOutOfOrderSequenceNumber:
  265. return "kafka server: The broker received an out of order sequence number."
  266. case ErrDuplicateSequenceNumber:
  267. return "kafka server: The broker received a duplicate sequence number."
  268. case ErrInvalidProducerEpoch:
  269. return "kafka server: Producer attempted an operation with an old epoch."
  270. case ErrInvalidTxnState:
  271. return "kafka server: The producer attempted a transactional operation in an invalid state."
  272. case ErrInvalidProducerIDMapping:
  273. return "kafka server: The producer attempted to use a producer id which is not currently assigned to its transactional id."
  274. case ErrInvalidTransactionTimeout:
  275. return "kafka server: The transaction timeout is larger than the maximum value allowed by the broker (as configured by max.transaction.timeout.ms)."
  276. case ErrConcurrentTransactions:
  277. return "kafka server: The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing."
  278. case ErrTransactionCoordinatorFenced:
  279. return "kafka server: The transaction coordinator sending a WriteTxnMarker is no longer the current coordinator for a given producer."
  280. case ErrTransactionalIDAuthorizationFailed:
  281. return "kafka server: Transactional ID authorization failed."
  282. case ErrSecurityDisabled:
  283. return "kafka server: Security features are disabled."
  284. case ErrOperationNotAttempted:
  285. return "kafka server: The broker did not attempt to execute this operation."
  286. case ErrKafkaStorageError:
  287. return "kafka server: Disk error when trying to access log file on the disk."
  288. case ErrLogDirNotFound:
  289. return "kafka server: The specified log directory is not found in the broker config."
  290. case ErrSASLAuthenticationFailed:
  291. return "kafka server: SASL Authentication failed."
  292. case ErrUnknownProducerID:
  293. return "kafka server: The broker could not locate the producer metadata associated with the Producer ID."
  294. case ErrReassignmentInProgress:
  295. return "kafka server: A partition reassignment is in progress."
  296. case ErrDelegationTokenAuthDisabled:
  297. return "kafka server: Delegation Token feature is not enabled."
  298. case ErrDelegationTokenNotFound:
  299. return "kafka server: Delegation Token is not found on server."
  300. case ErrDelegationTokenOwnerMismatch:
  301. return "kafka server: Specified Principal is not valid Owner/Renewer."
  302. case ErrDelegationTokenRequestNotAllowed:
  303. return "kafka server: Delegation Token requests are not allowed on PLAINTEXT/1-way SSL channels and on delegation token authenticated channels."
  304. case ErrDelegationTokenAuthorizationFailed:
  305. return "kafka server: Delegation Token authorization failed."
  306. case ErrDelegationTokenExpired:
  307. return "kafka server: Delegation Token is expired."
  308. case ErrInvalidPrincipalType:
  309. return "kafka server: Supplied principalType is not supported."
  310. case ErrNonEmptyGroup:
  311. return "kafka server: The group is not empty."
  312. case ErrGroupIDNotFound:
  313. return "kafka server: The group id does not exist."
  314. case ErrFetchSessionIDNotFound:
  315. return "kafka server: The fetch session ID was not found."
  316. case ErrInvalidFetchSessionEpoch:
  317. return "kafka server: The fetch session epoch is invalid."
  318. case ErrListenerNotFound:
  319. return "kafka server: There is no listener on the leader broker that matches the listener on which metadata request was processed."
  320. case ErrTopicDeletionDisabled:
  321. return "kafka server: Topic deletion is disabled."
  322. case ErrFencedLeaderEpoch:
  323. return "kafka server: The leader epoch in the request is older than the epoch on the broker."
  324. case ErrUnknownLeaderEpoch:
  325. return "kafka server: The leader epoch in the request is newer than the epoch on the broker."
  326. case ErrUnsupportedCompressionType:
  327. return "kafka server: The requesting client does not support the compression type of given partition."
  328. case ErrStaleBrokerEpoch:
  329. return "kafka server: Broker epoch has changed"
  330. case ErrOffsetNotAvailable:
  331. return "kafka server: The leader high watermark has not caught up from a recent leader election so the offsets cannot be guaranteed to be monotonically increasing"
  332. case ErrMemberIdRequired:
  333. return "kafka server: The group member needs to have a valid member id before actually entering a consumer group"
  334. case ErrPreferredLeaderNotAvailable:
  335. return "kafka server: The preferred leader was not available"
  336. case ErrGroupMaxSizeReached:
  337. return "kafka server: Consumer group The consumer group has reached its max size. already has the configured maximum number of members."
  338. case ErrFencedInstancedId:
  339. return "kafka server: The broker rejected this static consumer since another consumer with the same group.instance.id has registered with a different member.id."
  340. }
  341. return fmt.Sprintf("Unknown error, how did this happen? Error code = %d", err)
  342. }