| Abhay Kumar | a2ae599 | 2025-11-10 14:02:24 +0000 | [diff] [blame^] | 1 | package sarama |
| 2 | |
| 3 | import ( |
| 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. |
| 11 | var 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. |
| 14 | var 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. |
| 17 | var 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. |
| 21 | var 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]). |
| 25 | var 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. |
| 28 | var 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. |
| 31 | var 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. |
| 36 | var 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. |
| 39 | var 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 |
| 42 | var 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. |
| 46 | var 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. |
| 50 | var 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. |
| 54 | var ErrNoTopicsToUpdateMetadata = errors.New("kafka: no specific topics to update metadata") |
| 55 | |
| 56 | // ErrUnknownScramMechanism is returned when user tries to AlterUserScramCredentials with unknown SCRAM mechanism |
| 57 | var ErrUnknownScramMechanism = errors.New("kafka: unknown SCRAM mechanism provided") |
| 58 | |
| 59 | // ErrReassignPartitions is returned when altering partition assignments for a topic fails |
| 60 | var 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 |
| 63 | var ErrDeleteRecords = errors.New("kafka server: failed to delete records") |
| 64 | |
| 65 | // ErrCreateACLs is the type of error returned when ACL creation failed |
| 66 | var ErrCreateACLs = errors.New("kafka server: failed to create one or more ACL rules") |
| 67 | |
| 68 | // ErrAddPartitionsToTxn is returned when AddPartitionsToTxn failed multiple times |
| 69 | var ErrAddPartitionsToTxn = errors.New("transaction manager: failed to send partitions to transaction") |
| 70 | |
| 71 | // ErrTxnOffsetCommit is returned when TxnOffsetCommit failed multiple times |
| 72 | var ErrTxnOffsetCommit = errors.New("transaction manager: failed to send offsets to transaction") |
| 73 | |
| 74 | // ErrTransactionNotReady when transaction status is invalid for the current action. |
| 75 | var ErrTransactionNotReady = errors.New("transaction manager: transaction is not ready") |
| 76 | |
| 77 | // ErrNonTransactedProducer when calling BeginTxn, CommitTxn or AbortTxn on a non transactional producer. |
| 78 | var ErrNonTransactedProducer = errors.New("transaction manager: you need to add TransactionalID to producer") |
| 79 | |
| 80 | // ErrTransitionNotAllowed when txnmgr state transition is not valid. |
| 81 | var ErrTransitionNotAllowed = errors.New("transaction manager: invalid transition attempted") |
| 82 | |
| 83 | // ErrCannotTransitionNilError when transition is attempted with an nil error. |
| 84 | var ErrCannotTransitionNilError = errors.New("transaction manager: cannot transition with a nil error") |
| 85 | |
| 86 | // ErrTxnUnableToParseResponse when response is nil |
| 87 | var ErrTxnUnableToParseResponse = errors.New("transaction manager: unable to parse response") |
| 88 | |
| 89 | // ErrUnknownMessage when the protocol message key is not recognized |
| 90 | var 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. |
| 95 | func 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 | |
| 110 | type sentinelError struct { |
| 111 | sentinel error |
| 112 | wrapped error |
| 113 | } |
| 114 | |
| 115 | func (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 | |
| 123 | func (err sentinelError) Is(target error) bool { |
| 124 | return errors.Is(err.sentinel, target) || errors.Is(err.wrapped, target) |
| 125 | } |
| 126 | |
| 127 | func (err sentinelError) Unwrap() error { |
| 128 | return err.wrapped |
| 129 | } |
| 130 | |
| 131 | func 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. |
| 137 | type PacketEncodingError struct { |
| 138 | Info string |
| 139 | } |
| 140 | |
| 141 | func (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. |
| 147 | type PacketDecodingError struct { |
| 148 | Info string |
| 149 | } |
| 150 | |
| 151 | func (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. |
| 157 | type ConfigurationError string |
| 158 | |
| 159 | func (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 |
| 165 | type KError int16 |
| 166 | |
| 167 | // Numeric error codes returned by the Kafka server. |
| 168 | const ( |
| 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 | |
| 263 | func (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 | } |