# frozen_string_literal: true require "kafka/version" module Kafka class Error < StandardError end # There was an error processing a message. class ProcessingError < Error attr_reader :topic, :partition, :offset def initialize(topic, partition, offset) @topic = topic @partition = partition @offset = offset super() end end # Subclasses of this exception class map to an error code described in the # Kafka protocol specification. # https://kafka.apache.org/protocol#protocol_error_codes class ProtocolError < Error end # -1 # The server experienced an unexpected error when processing the request class UnknownError < ProtocolError end # 1 # The requested offset is not within the range of offsets maintained by the server. class OffsetOutOfRange < ProtocolError attr_accessor :topic, :partition, :offset end # 2 # This indicates that a message contents does not match its CRC. class CorruptMessage < ProtocolError end # 3 # The request is for a topic or partition that does not exist on the broker. class UnknownTopicOrPartition < ProtocolError end # 4 # The message has a negative size. class InvalidMessageSize < ProtocolError end # 5 # This error is thrown if we are in the middle of a leadership election and # there is currently no leader for this partition and hence it is unavailable # for writes. class LeaderNotAvailable < ProtocolError end # 6 # This error is thrown if the client attempts to send messages to a replica # that is not the leader for some partition. It indicates that the client's # metadata is out of date. class NotLeaderForPartition < ProtocolError end # 7 # This error is thrown if the request exceeds the user-specified time limit # in the request. class RequestTimedOut < ProtocolError end # 8 # The broker is not available. class BrokerNotAvailable < ProtocolError end # 9 # Raised if a replica is expected on a broker, but is not. Can be safely ignored. class ReplicaNotAvailable < ProtocolError end # 10 # The server has a configurable maximum message size to avoid unbounded memory # allocation. This error is thrown if the client attempt to produce a message # larger than this maximum. class MessageSizeTooLarge < ProtocolError end # 11 # The controller moved to another broker. class StaleControllerEpoch < ProtocolError end # 12 # If you specify a string larger than configured maximum for offset metadata. class OffsetMetadataTooLarge < ProtocolError end # 13 # The server disconnected before a response was received. class NetworkException < ProtocolError end # 14 # The coordinator is loading and hence can't process requests. class CoordinatorLoadInProgress < ProtocolError end # 15 # The coordinator is not available. class CoordinatorNotAvailable < ProtocolError end # 16 # This is not the correct coordinator. class NotCoordinatorForGroup < ProtocolError end # 17 # For a request which attempts to access an invalid topic (e.g. one which has # an illegal name), or if an attempt is made to write to an internal topic # (such as the consumer offsets topic). class InvalidTopic < ProtocolError end # 18 # If a message batch in a produce request exceeds the maximum configured # segment size. class RecordListTooLarge < ProtocolError end # 19 # Returned from a produce request when the number of in-sync replicas is # lower than the configured minimum and requiredAcks is -1. class NotEnoughReplicas < ProtocolError end # 20 # Returned from a produce request when the message was written to the log, # but with fewer in-sync replicas than required. class NotEnoughReplicasAfterAppend < ProtocolError end # 21 # Returned from a produce request if the requested requiredAcks is invalid # (anything other than -1, 1, or 0). class InvalidRequiredAcks < ProtocolError end # 22 # Specified group generation id is not valid. class IllegalGeneration < ProtocolError end # 23 # The group member's supported protocols are incompatible with those of existing members or first group member tried to join with empty protocol type or empty protocol list. class InconsistentGroupProtocol < ProtocolError end # 24 # The configured groupId is invalid class InvalidGroupId < ProtocolError end # 25 # The coordinator is not aware of this member. class UnknownMemberId < ProtocolError end # 26 # The session timeout is not within the range allowed by the broker class InvalidSessionTimeout < ProtocolError end # 27 # The group is rebalancing, so a rejoin is needed. class RebalanceInProgress < ProtocolError end # 28 # The committing offset data size is not valid class InvalidCommitOffsetSize < ProtocolError end # 29 class TopicAuthorizationFailed < ProtocolError end # 30 class GroupAuthorizationFailed < ProtocolError end # 31 class ClusterAuthorizationFailed < ProtocolError end # 32 # The timestamp of the message is out of acceptable range. class InvalidTimestamp < ProtocolError end # 33 # The broker does not support the requested SASL mechanism. class UnsupportedSaslMechanism < ProtocolError end # 34 class InvalidSaslState < ProtocolError end # 35 class UnsupportedVersion < ProtocolError end # 36 class TopicAlreadyExists < ProtocolError end # 37 # Number of partitions is below 1. class InvalidPartitions < ProtocolError end # 38 # Replication factor is below 1 or larger than the number of available brokers. class InvalidReplicationFactor < ProtocolError end # 39 class InvalidReplicaAssignment < ProtocolError end # 40 class InvalidConfig < ProtocolError end # 41 # This is not the correct controller for this cluster. class NotController < ProtocolError end # 42 class InvalidRequest < ProtocolError end # 43 # The message format version on the broker does not support the request. class UnsupportedForMessageFormat < ProtocolError end # 44 # Request parameters do not satisfy the configured policy. class PolicyViolation < ProtocolError end # 45 # The broker received an out of order sequence number class OutOfOrderSequenceNumberError < Error end # 46 # The broker received a duplicate sequence number class DuplicateSequenceNumberError < Error end # 47 # Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer's transaction has been expired by the broker. class InvalidProducerEpochError < Error end # 48 # The producer attempted a transactional operation in an invalid state class InvalidTxnStateError < Error end # 49 # The producer attempted to use a producer id which is not currently assigned to its transactional id class InvalidProducerIDMappingError < Error end # 50 # The transaction timeout is larger than the maximum value allowed by the broker (as configured by transaction.max.timeout.ms). class InvalidTransactionTimeoutError < Error end # 51 # The producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing class ConcurrentTransactionError < Error end # 52 # Indicates that the transaction coordinator sending a WriteTxnMarker is no longer the current coordinator for a given producer class TransactionCoordinatorFencedError < Error end ### # ruby-kafka errors ### # A fetch operation was executed with no partitions specified. class NoPartitionsToFetchFrom < Error end # A message in a partition is larger than the maximum we've asked for. class MessageTooLargeToRead < Error end # A connection has been unused for too long, we assume the server has killed it. class IdleConnection < Error end # When the record array length doesn't match real number of received records class InsufficientDataMessage < Error end # Raised when there's a network connection error. class ConnectionError < Error end class NoSuchBroker < Error end # Raised when a producer buffer has reached its maximum size. class BufferOverflow < Error end # Raised if not all messages could be sent by a producer. class DeliveryFailed < Error attr_reader :failed_messages def initialize(message, failed_messages) @failed_messages = failed_messages super(message) end end class HeartbeatError < Error end class OffsetCommitError < Error end class FetchError < Error end class SaslScramError < Error end class FailedScramAuthentication < SaslScramError end # The Token Provider object used for SASL OAuthBearer does not implement the method `token` class TokenMethodNotImplementedError < Error end # Initializes a new Kafka client. # # @see Client#initialize # @return [Client] def self.new(seed_brokers = nil, **options) # We allow `seed_brokers` to be passed in either as a positional _or_ as a # keyword argument. if seed_brokers.nil? Client.new(**options) else Client.new(seed_brokers: seed_brokers, **options) end end end require "kafka/client"