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