blob: 55af65307b29220f37ae25c25caaf31d48fdc28c [file] [log] [blame]
Abhay Kumara2ae5992025-11-10 14:02:24 +00001package sarama
2
3import (
4 "errors"
5 "fmt"
6 "strings"
7)
8
9// ErrOutOfBrokers is the error returned when the client has run out of brokers to talk to because all of them errored
10// or otherwise failed to respond.
11var ErrOutOfBrokers = errors.New("kafka: client has run out of available brokers to talk to")
12
13// ErrBrokerNotFound is the error returned when there's no broker found for the requested ID.
14var ErrBrokerNotFound = errors.New("kafka: broker for ID is not found")
15
16// ErrClosedClient is the error returned when a method is called on a client that has been closed.
17var ErrClosedClient = errors.New("kafka: tried to use a client that was closed")
18
19// ErrIncompleteResponse is the error returned when the server returns a syntactically valid response, but it does
20// not contain the expected information.
21var ErrIncompleteResponse = errors.New("kafka: response did not contain all the expected topic/partition blocks")
22
23// ErrInvalidPartition is the error returned when a partitioner returns an invalid partition index
24// (meaning one outside of the range [0...numPartitions-1]).
25var ErrInvalidPartition = errors.New("kafka: partitioner returned an invalid partition index")
26
27// ErrAlreadyConnected is the error returned when calling Open() on a Broker that is already connected or connecting.
28var ErrAlreadyConnected = errors.New("kafka: broker connection already initiated")
29
30// ErrNotConnected is the error returned when trying to send or call Close() on a Broker that is not connected.
31var ErrNotConnected = errors.New("kafka: broker not connected")
32
33// ErrInsufficientData is returned when decoding and the packet is truncated. This can be expected
34// when requesting messages, since as an optimization the server is allowed to return a partial message at the end
35// of the message set.
36var ErrInsufficientData = errors.New("kafka: insufficient data to decode packet, more bytes expected")
37
38// ErrShuttingDown is returned when a producer receives a message during shutdown.
39var ErrShuttingDown = errors.New("kafka: message received by producer in process of shutting down")
40
41// ErrMessageTooLarge is returned when the next message to consume is larger than the configured Consumer.Fetch.Max
42var ErrMessageTooLarge = errors.New("kafka: message is larger than Consumer.Fetch.Max")
43
44// ErrConsumerOffsetNotAdvanced is returned when a partition consumer didn't advance its offset after parsing
45// a RecordBatch.
46var ErrConsumerOffsetNotAdvanced = errors.New("kafka: consumer offset was not advanced after a RecordBatch")
47
48// ErrControllerNotAvailable is returned when server didn't give correct controller id. May be kafka server's version
49// is lower than 0.10.0.0.
50var ErrControllerNotAvailable = errors.New("kafka: controller is not available")
51
52// ErrNoTopicsToUpdateMetadata is returned when Meta.Full is set to false but no specific topics were found to update
53// the metadata.
54var ErrNoTopicsToUpdateMetadata = errors.New("kafka: no specific topics to update metadata")
55
56// ErrUnknownScramMechanism is returned when user tries to AlterUserScramCredentials with unknown SCRAM mechanism
57var ErrUnknownScramMechanism = errors.New("kafka: unknown SCRAM mechanism provided")
58
59// ErrReassignPartitions is returned when altering partition assignments for a topic fails
60var ErrReassignPartitions = errors.New("failed to reassign partitions for topic")
61
62// ErrDeleteRecords is the type of error returned when fail to delete the required records
63var ErrDeleteRecords = errors.New("kafka server: failed to delete records")
64
65// ErrCreateACLs is the type of error returned when ACL creation failed
66var ErrCreateACLs = errors.New("kafka server: failed to create one or more ACL rules")
67
68// ErrAddPartitionsToTxn is returned when AddPartitionsToTxn failed multiple times
69var ErrAddPartitionsToTxn = errors.New("transaction manager: failed to send partitions to transaction")
70
71// ErrTxnOffsetCommit is returned when TxnOffsetCommit failed multiple times
72var ErrTxnOffsetCommit = errors.New("transaction manager: failed to send offsets to transaction")
73
74// ErrTransactionNotReady when transaction status is invalid for the current action.
75var ErrTransactionNotReady = errors.New("transaction manager: transaction is not ready")
76
77// ErrNonTransactedProducer when calling BeginTxn, CommitTxn or AbortTxn on a non transactional producer.
78var ErrNonTransactedProducer = errors.New("transaction manager: you need to add TransactionalID to producer")
79
80// ErrTransitionNotAllowed when txnmgr state transition is not valid.
81var ErrTransitionNotAllowed = errors.New("transaction manager: invalid transition attempted")
82
83// ErrCannotTransitionNilError when transition is attempted with an nil error.
84var ErrCannotTransitionNilError = errors.New("transaction manager: cannot transition with a nil error")
85
86// ErrTxnUnableToParseResponse when response is nil
87var ErrTxnUnableToParseResponse = errors.New("transaction manager: unable to parse response")
88
89// ErrUnknownMessage when the protocol message key is not recognized
90var ErrUnknownMessage = errors.New("kafka: unknown protocol message key")
91
92// MultiErrorFormat specifies the formatter applied to format multierrors.
93//
94// Deprecated: Please use [errors.Join] instead.
95func MultiErrorFormat(es []error) string {
96 if len(es) == 1 {
97 return es[0].Error()
98 }
99
100 points := make([]string, len(es))
101 for i, err := range es {
102 points[i] = fmt.Sprintf("* %s", err)
103 }
104
105 return fmt.Sprintf(
106 "%d errors occurred:\n\t%s\n",
107 len(es), strings.Join(points, "\n\t"))
108}
109
110type sentinelError struct {
111 sentinel error
112 wrapped error
113}
114
115func (err sentinelError) Error() string {
116 if err.wrapped != nil {
117 return fmt.Sprintf("%s: %v", err.sentinel, err.wrapped)
118 } else {
119 return fmt.Sprintf("%s", err.sentinel)
120 }
121}
122
123func (err sentinelError) Is(target error) bool {
124 return errors.Is(err.sentinel, target) || errors.Is(err.wrapped, target)
125}
126
127func (err sentinelError) Unwrap() error {
128 return err.wrapped
129}
130
131func Wrap(sentinel error, wrapped ...error) sentinelError {
132 return sentinelError{sentinel: sentinel, wrapped: errors.Join(wrapped...)}
133}
134
135// PacketEncodingError is returned from a failure while encoding a Kafka packet. This can happen, for example,
136// if you try to encode a string over 2^15 characters in length, since Kafka's encoding rules do not permit that.
137type PacketEncodingError struct {
138 Info string
139}
140
141func (err PacketEncodingError) Error() string {
142 return fmt.Sprintf("kafka: error encoding packet: %s", err.Info)
143}
144
145// PacketDecodingError is returned when there was an error (other than truncated data) decoding the Kafka broker's response.
146// This can be a bad CRC or length field, or any other invalid value.
147type PacketDecodingError struct {
148 Info string
149}
150
151func (err PacketDecodingError) Error() string {
152 return fmt.Sprintf("kafka: error decoding packet: %s", err.Info)
153}
154
155// ConfigurationError is the type of error returned from a constructor (e.g. NewClient, or NewConsumer)
156// when the specified configuration is invalid.
157type ConfigurationError string
158
159func (err ConfigurationError) Error() string {
160 return "kafka: invalid configuration (" + string(err) + ")"
161}
162
163// KError is the type of error that can be returned directly by the Kafka broker.
164// See https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ErrorCodes
165type KError int16
166
167// Numeric error codes returned by the Kafka server.
168const (
169 ErrUnknown KError = -1 // Errors.UNKNOWN_SERVER_ERROR
170 ErrNoError KError = 0 // Errors.NONE
171 ErrOffsetOutOfRange KError = 1 // Errors.OFFSET_OUT_OF_RANGE
172 ErrInvalidMessage KError = 2 // Errors.CORRUPT_MESSAGE
173 ErrUnknownTopicOrPartition KError = 3 // Errors.UNKNOWN_TOPIC_OR_PARTITION
174 ErrInvalidMessageSize KError = 4 // Errors.INVALID_FETCH_SIZE
175 ErrLeaderNotAvailable KError = 5 // Errors.LEADER_NOT_AVAILABLE
176 ErrNotLeaderForPartition KError = 6 // Errors.NOT_LEADER_OR_FOLLOWER
177 ErrRequestTimedOut KError = 7 // Errors.REQUEST_TIMED_OUT
178 ErrBrokerNotAvailable KError = 8 // Errors.BROKER_NOT_AVAILABLE
179 ErrReplicaNotAvailable KError = 9 // Errors.REPLICA_NOT_AVAILABLE
180 ErrMessageSizeTooLarge KError = 10 // Errors.MESSAGE_TOO_LARGE
181 ErrStaleControllerEpochCode KError = 11 // Errors.STALE_CONTROLLER_EPOCH
182 ErrOffsetMetadataTooLarge KError = 12 // Errors.OFFSET_METADATA_TOO_LARGE
183 ErrNetworkException KError = 13 // Errors.NETWORK_EXCEPTION
184 ErrOffsetsLoadInProgress KError = 14 // Errors.COORDINATOR_LOAD_IN_PROGRESS
185 ErrConsumerCoordinatorNotAvailable KError = 15 // Errors.COORDINATOR_NOT_AVAILABLE
186 ErrNotCoordinatorForConsumer KError = 16 // Errors.NOT_COORDINATOR
187 ErrInvalidTopic KError = 17 // Errors.INVALID_TOPIC_EXCEPTION
188 ErrMessageSetSizeTooLarge KError = 18 // Errors.RECORD_LIST_TOO_LARGE
189 ErrNotEnoughReplicas KError = 19 // Errors.NOT_ENOUGH_REPLICAS
190 ErrNotEnoughReplicasAfterAppend KError = 20 // Errors.NOT_ENOUGH_REPLICAS_AFTER_APPEND
191 ErrInvalidRequiredAcks KError = 21 // Errors.INVALID_REQUIRED_ACKS
192 ErrIllegalGeneration KError = 22 // Errors.ILLEGAL_GENERATION
193 ErrInconsistentGroupProtocol KError = 23 // Errors.INCONSISTENT_GROUP_PROTOCOL
194 ErrInvalidGroupId KError = 24 // Errors.INVALID_GROUP_ID
195 ErrUnknownMemberId KError = 25 // Errors.UNKNOWN_MEMBER_ID
196 ErrInvalidSessionTimeout KError = 26 // Errors.INVALID_SESSION_TIMEOUT
197 ErrRebalanceInProgress KError = 27 // Errors.REBALANCE_IN_PROGRESS
198 ErrInvalidCommitOffsetSize KError = 28 // Errors.INVALID_COMMIT_OFFSET_SIZE
199 ErrTopicAuthorizationFailed KError = 29 // Errors.TOPIC_AUTHORIZATION_FAILED
200 ErrGroupAuthorizationFailed KError = 30 // Errors.GROUP_AUTHORIZATION_FAILED
201 ErrClusterAuthorizationFailed KError = 31 // Errors.CLUSTER_AUTHORIZATION_FAILED
202 ErrInvalidTimestamp KError = 32 // Errors.INVALID_TIMESTAMP
203 ErrUnsupportedSASLMechanism KError = 33 // Errors.UNSUPPORTED_SASL_MECHANISM
204 ErrIllegalSASLState KError = 34 // Errors.ILLEGAL_SASL_STATE
205 ErrUnsupportedVersion KError = 35 // Errors.UNSUPPORTED_VERSION
206 ErrTopicAlreadyExists KError = 36 // Errors.TOPIC_ALREADY_EXISTS
207 ErrInvalidPartitions KError = 37 // Errors.INVALID_PARTITIONS
208 ErrInvalidReplicationFactor KError = 38 // Errors.INVALID_REPLICATION_FACTOR
209 ErrInvalidReplicaAssignment KError = 39 // Errors.INVALID_REPLICA_ASSIGNMENT
210 ErrInvalidConfig KError = 40 // Errors.INVALID_CONFIG
211 ErrNotController KError = 41 // Errors.NOT_CONTROLLER
212 ErrInvalidRequest KError = 42 // Errors.INVALID_REQUEST
213 ErrUnsupportedForMessageFormat KError = 43 // Errors.UNSUPPORTED_FOR_MESSAGE_FORMAT
214 ErrPolicyViolation KError = 44 // Errors.POLICY_VIOLATION
215 ErrOutOfOrderSequenceNumber KError = 45 // Errors.OUT_OF_ORDER_SEQUENCE_NUMBER
216 ErrDuplicateSequenceNumber KError = 46 // Errors.DUPLICATE_SEQUENCE_NUMBER
217 ErrInvalidProducerEpoch KError = 47 // Errors.INVALID_PRODUCER_EPOCH
218 ErrInvalidTxnState KError = 48 // Errors.INVALID_TXN_STATE
219 ErrInvalidProducerIDMapping KError = 49 // Errors.INVALID_PRODUCER_ID_MAPPING
220 ErrInvalidTransactionTimeout KError = 50 // Errors.INVALID_TRANSACTION_TIMEOUT
221 ErrConcurrentTransactions KError = 51 // Errors.CONCURRENT_TRANSACTIONS
222 ErrTransactionCoordinatorFenced KError = 52 // Errors.TRANSACTION_COORDINATOR_FENCED
223 ErrTransactionalIDAuthorizationFailed KError = 53 // Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED
224 ErrSecurityDisabled KError = 54 // Errors.SECURITY_DISABLED
225 ErrOperationNotAttempted KError = 55 // Errors.OPERATION_NOT_ATTEMPTED
226 ErrKafkaStorageError KError = 56 // Errors.KAFKA_STORAGE_ERROR
227 ErrLogDirNotFound KError = 57 // Errors.LOG_DIR_NOT_FOUND
228 ErrSASLAuthenticationFailed KError = 58 // Errors.SASL_AUTHENTICATION_FAILED
229 ErrUnknownProducerID KError = 59 // Errors.UNKNOWN_PRODUCER_ID
230 ErrReassignmentInProgress KError = 60 // Errors.REASSIGNMENT_IN_PROGRESS
231 ErrDelegationTokenAuthDisabled KError = 61 // Errors.DELEGATION_TOKEN_AUTH_DISABLED
232 ErrDelegationTokenNotFound KError = 62 // Errors.DELEGATION_TOKEN_NOT_FOUND
233 ErrDelegationTokenOwnerMismatch KError = 63 // Errors.DELEGATION_TOKEN_OWNER_MISMATCH
234 ErrDelegationTokenRequestNotAllowed KError = 64 // Errors.DELEGATION_TOKEN_REQUEST_NOT_ALLOWED
235 ErrDelegationTokenAuthorizationFailed KError = 65 // Errors.DELEGATION_TOKEN_AUTHORIZATION_FAILED
236 ErrDelegationTokenExpired KError = 66 // Errors.DELEGATION_TOKEN_EXPIRED
237 ErrInvalidPrincipalType KError = 67 // Errors.INVALID_PRINCIPAL_TYPE
238 ErrNonEmptyGroup KError = 68 // Errors.NON_EMPTY_GROUP
239 ErrGroupIDNotFound KError = 69 // Errors.GROUP_ID_NOT_FOUND
240 ErrFetchSessionIDNotFound KError = 70 // Errors.FETCH_SESSION_ID_NOT_FOUND
241 ErrInvalidFetchSessionEpoch KError = 71 // Errors.INVALID_FETCH_SESSION_EPOCH
242 ErrListenerNotFound KError = 72 // Errors.LISTENER_NOT_FOUND
243 ErrTopicDeletionDisabled KError = 73 // Errors.TOPIC_DELETION_DISABLED
244 ErrFencedLeaderEpoch KError = 74 // Errors.FENCED_LEADER_EPOCH
245 ErrUnknownLeaderEpoch KError = 75 // Errors.UNKNOWN_LEADER_EPOCH
246 ErrUnsupportedCompressionType KError = 76 // Errors.UNSUPPORTED_COMPRESSION_TYPE
247 ErrStaleBrokerEpoch KError = 77 // Errors.STALE_BROKER_EPOCH
248 ErrOffsetNotAvailable KError = 78 // Errors.OFFSET_NOT_AVAILABLE
249 ErrMemberIdRequired KError = 79 // Errors.MEMBER_ID_REQUIRED
250 ErrPreferredLeaderNotAvailable KError = 80 // Errors.PREFERRED_LEADER_NOT_AVAILABLE
251 ErrGroupMaxSizeReached KError = 81 // Errors.GROUP_MAX_SIZE_REACHED
252 ErrFencedInstancedId KError = 82 // Errors.FENCED_INSTANCE_ID
253 ErrEligibleLeadersNotAvailable KError = 83 // Errors.ELIGIBLE_LEADERS_NOT_AVAILABLE
254 ErrElectionNotNeeded KError = 84 // Errors.ELECTION_NOT_NEEDED
255 ErrNoReassignmentInProgress KError = 85 // Errors.NO_REASSIGNMENT_IN_PROGRESS
256 ErrGroupSubscribedToTopic KError = 86 // Errors.GROUP_SUBSCRIBED_TO_TOPIC
257 ErrInvalidRecord KError = 87 // Errors.INVALID_RECORD
258 ErrUnstableOffsetCommit KError = 88 // Errors.UNSTABLE_OFFSET_COMMIT
259 ErrThrottlingQuotaExceeded KError = 89 // Errors.THROTTLING_QUOTA_EXCEEDED
260 ErrProducerFenced KError = 90 // Errors.PRODUCER_FENCED
261)
262
263func (err KError) Error() string {
264 // Error messages stolen/adapted from
265 // https://kafka.apache.org/protocol#protocol_error_codes
266 switch err {
267 case ErrNoError:
268 return "kafka server: Not an error, why are you printing me?"
269 case ErrUnknown:
270 return "kafka server: Unexpected (unknown?) server error"
271 case ErrOffsetOutOfRange:
272 return "kafka server: The requested offset is outside the range of offsets maintained by the server for the given topic/partition"
273 case ErrInvalidMessage:
274 return "kafka server: Message contents does not match its CRC"
275 case ErrUnknownTopicOrPartition:
276 return "kafka server: Request was for a topic or partition that does not exist on this broker"
277 case ErrInvalidMessageSize:
278 return "kafka server: The message has a negative size"
279 case ErrLeaderNotAvailable:
280 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"
281 case ErrNotLeaderForPartition:
282 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"
283 case ErrRequestTimedOut:
284 return "kafka server: Request exceeded the user-specified time limit in the request"
285 case ErrBrokerNotAvailable:
286 return "kafka server: Broker not available. Not a client facing error, we should never receive this!!!"
287 case ErrReplicaNotAvailable:
288 return "kafka server: Replica information not available, one or more brokers are down"
289 case ErrMessageSizeTooLarge:
290 return "kafka server: Message was too large, server rejected it to avoid allocation error"
291 case ErrStaleControllerEpochCode:
292 return "kafka server: StaleControllerEpochCode (internal error code for broker-to-broker communication)"
293 case ErrOffsetMetadataTooLarge:
294 return "kafka server: Specified a string larger than the configured maximum for offset metadata"
295 case ErrNetworkException:
296 return "kafka server: The server disconnected before a response was received"
297 case ErrOffsetsLoadInProgress:
298 return "kafka server: The coordinator is still loading offsets and cannot currently process requests"
299 case ErrConsumerCoordinatorNotAvailable:
300 return "kafka server: The coordinator is not available"
301 case ErrNotCoordinatorForConsumer:
302 return "kafka server: Request was for a consumer group that is not coordinated by this broker"
303 case ErrInvalidTopic:
304 return "kafka server: The request attempted to perform an operation on an invalid topic"
305 case ErrMessageSetSizeTooLarge:
306 return "kafka server: The request included message batch larger than the configured segment size on the server"
307 case ErrNotEnoughReplicas:
308 return "kafka server: Messages are rejected since there are fewer in-sync replicas than required"
309 case ErrNotEnoughReplicasAfterAppend:
310 return "kafka server: Messages are written to the log, but to fewer in-sync replicas than required"
311 case ErrInvalidRequiredAcks:
312 return "kafka server: The number of required acks is invalid (should be either -1, 0, or 1)"
313 case ErrIllegalGeneration:
314 return "kafka server: The provided generation id is not the current generation"
315 case ErrInconsistentGroupProtocol:
316 return "kafka server: The provider group protocol type is incompatible with the other members"
317 case ErrInvalidGroupId:
318 return "kafka server: The provided group id was empty"
319 case ErrUnknownMemberId:
320 return "kafka server: The provided member is not known in the current generation"
321 case ErrInvalidSessionTimeout:
322 return "kafka server: The provided session timeout is outside the allowed range"
323 case ErrRebalanceInProgress:
324 return "kafka server: A rebalance for the group is in progress. Please re-join the group"
325 case ErrInvalidCommitOffsetSize:
326 return "kafka server: The provided commit metadata was too large"
327 case ErrTopicAuthorizationFailed:
328 return "kafka server: The client is not authorized to access this topic"
329 case ErrGroupAuthorizationFailed:
330 return "kafka server: The client is not authorized to access this group"
331 case ErrClusterAuthorizationFailed:
332 return "kafka server: The client is not authorized to send this request type"
333 case ErrInvalidTimestamp:
334 return "kafka server: The timestamp of the message is out of acceptable range"
335 case ErrUnsupportedSASLMechanism:
336 return "kafka server: The broker does not support the requested SASL mechanism"
337 case ErrIllegalSASLState:
338 return "kafka server: Request is not valid given the current SASL state"
339 case ErrUnsupportedVersion:
340 return "kafka server: The version of API is not supported"
341 case ErrTopicAlreadyExists:
342 return "kafka server: Topic with this name already exists"
343 case ErrInvalidPartitions:
344 return "kafka server: Number of partitions is invalid"
345 case ErrInvalidReplicationFactor:
346 return "kafka server: Replication-factor is invalid"
347 case ErrInvalidReplicaAssignment:
348 return "kafka server: Replica assignment is invalid"
349 case ErrInvalidConfig:
350 return "kafka server: Configuration is invalid"
351 case ErrNotController:
352 return "kafka server: This is not the correct controller for this cluster"
353 case ErrInvalidRequest:
354 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"
355 case ErrUnsupportedForMessageFormat:
356 return "kafka server: The requested operation is not supported by the message format version"
357 case ErrPolicyViolation:
358 return "kafka server: Request parameters do not satisfy the configured policy"
359 case ErrOutOfOrderSequenceNumber:
360 return "kafka server: The broker received an out of order sequence number"
361 case ErrDuplicateSequenceNumber:
362 return "kafka server: The broker received a duplicate sequence number"
363 case ErrInvalidProducerEpoch:
364 return "kafka server: Producer attempted an operation with an old epoch"
365 case ErrInvalidTxnState:
366 return "kafka server: The producer attempted a transactional operation in an invalid state"
367 case ErrInvalidProducerIDMapping:
368 return "kafka server: The producer attempted to use a producer id which is not currently assigned to its transactional id"
369 case ErrInvalidTransactionTimeout:
370 return "kafka server: The transaction timeout is larger than the maximum value allowed by the broker (as configured by max.transaction.timeout.ms)"
371 case ErrConcurrentTransactions:
372 return "kafka server: The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing"
373 case ErrTransactionCoordinatorFenced:
374 return "kafka server: The transaction coordinator sending a WriteTxnMarker is no longer the current coordinator for a given producer"
375 case ErrTransactionalIDAuthorizationFailed:
376 return "kafka server: Transactional ID authorization failed"
377 case ErrSecurityDisabled:
378 return "kafka server: Security features are disabled"
379 case ErrOperationNotAttempted:
380 return "kafka server: The broker did not attempt to execute this operation"
381 case ErrKafkaStorageError:
382 return "kafka server: Disk error when trying to access log file on the disk"
383 case ErrLogDirNotFound:
384 return "kafka server: The specified log directory is not found in the broker config"
385 case ErrSASLAuthenticationFailed:
386 return "kafka server: SASL Authentication failed"
387 case ErrUnknownProducerID:
388 return "kafka server: The broker could not locate the producer metadata associated with the Producer ID"
389 case ErrReassignmentInProgress:
390 return "kafka server: A partition reassignment is in progress"
391 case ErrDelegationTokenAuthDisabled:
392 return "kafka server: Delegation Token feature is not enabled"
393 case ErrDelegationTokenNotFound:
394 return "kafka server: Delegation Token is not found on server"
395 case ErrDelegationTokenOwnerMismatch:
396 return "kafka server: Specified Principal is not valid Owner/Renewer"
397 case ErrDelegationTokenRequestNotAllowed:
398 return "kafka server: Delegation Token requests are not allowed on PLAINTEXT/1-way SSL channels and on delegation token authenticated channels"
399 case ErrDelegationTokenAuthorizationFailed:
400 return "kafka server: Delegation Token authorization failed"
401 case ErrDelegationTokenExpired:
402 return "kafka server: Delegation Token is expired"
403 case ErrInvalidPrincipalType:
404 return "kafka server: Supplied principalType is not supported"
405 case ErrNonEmptyGroup:
406 return "kafka server: The group is not empty"
407 case ErrGroupIDNotFound:
408 return "kafka server: The group id does not exist"
409 case ErrFetchSessionIDNotFound:
410 return "kafka server: The fetch session ID was not found"
411 case ErrInvalidFetchSessionEpoch:
412 return "kafka server: The fetch session epoch is invalid"
413 case ErrListenerNotFound:
414 return "kafka server: There is no listener on the leader broker that matches the listener on which metadata request was processed"
415 case ErrTopicDeletionDisabled:
416 return "kafka server: Topic deletion is disabled"
417 case ErrFencedLeaderEpoch:
418 return "kafka server: The leader epoch in the request is older than the epoch on the broker"
419 case ErrUnknownLeaderEpoch:
420 return "kafka server: The leader epoch in the request is newer than the epoch on the broker"
421 case ErrUnsupportedCompressionType:
422 return "kafka server: The requesting client does not support the compression type of given partition"
423 case ErrStaleBrokerEpoch:
424 return "kafka server: Broker epoch has changed"
425 case ErrOffsetNotAvailable:
426 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"
427 case ErrMemberIdRequired:
428 return "kafka server: The group member needs to have a valid member id before actually entering a consumer group"
429 case ErrPreferredLeaderNotAvailable:
430 return "kafka server: The preferred leader was not available"
431 case ErrGroupMaxSizeReached:
432 return "kafka server: Consumer group The consumer group has reached its max size. already has the configured maximum number of members"
433 case ErrFencedInstancedId:
434 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"
435 case ErrEligibleLeadersNotAvailable:
436 return "kafka server: Eligible topic partition leaders are not available"
437 case ErrElectionNotNeeded:
438 return "kafka server: Leader election not needed for topic partition"
439 case ErrNoReassignmentInProgress:
440 return "kafka server: No partition reassignment is in progress"
441 case ErrGroupSubscribedToTopic:
442 return "kafka server: Deleting offsets of a topic is forbidden while the consumer group is actively subscribed to it"
443 case ErrInvalidRecord:
444 return "kafka server: This record has failed the validation on broker and hence will be rejected"
445 case ErrUnstableOffsetCommit:
446 return "kafka server: There are unstable offsets that need to be cleared"
447 }
448
449 return fmt.Sprintf("Unknown error, how did this happen? Error code = %d", err)
450}