Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

kafka

package

Go to main page

Versions in this module

v0

v0.12.0

Aug 7, 2024

v0.11.4

Sep 14, 2024

v0.11.3

Aug 8, 2024

v0.11.2

Aug 6, 2024

v0.11.1

Aug 5, 2024

v0.11.0

Jul 24, 2024

v0.10.0

Jul 14, 2024

Changes in this version

+ const MaxRequestSize

+ const MaxResponseSize

+ var DefaultVersion = V2_1_0_0

+ var ErrAddPartitionsToTxn = errors.New("transaction manager: failed to send partitions to transaction")

+ var ErrAlreadyConnected = errors.New("kafka: broker connection already initiated")

+ var ErrBrokerNotFound = errors.New("kafka: broker for ID is not found")

+ var ErrCannotTransitionNilError = errors.New("transaction manager: cannot transition with a nil error")

+ var ErrClosedClient = errors.New("kafka: tried to use a client that was closed")

+ var ErrConsumerOffsetNotAdvanced = errors.New("kafka: consumer offset was not advanced after a RecordBatch")

+ var ErrControllerNotAvailable = errors.New("kafka: controller is not available")

+ var ErrCreateACLs = errors.New("kafka server: failed to create one or more ACL rules")

+ var ErrDeleteRecords = errors.New("kafka server: failed to delete records")

+ var ErrIncompleteResponse = errors.New("kafka: response did not contain all the expected topic/partition blocks")

+ var ErrInsufficientData = errors.New("kafka: insufficient data to decode packet, more bytes expected")

+ var ErrInvalidPartition = errors.New("kafka: partitioner returned an invalid partition index")

+ var ErrMessageTooLarge = errors.New("kafka: message is larger than Consumer.Fetch.Max")

+ var ErrNoTopicsToUpdateMetadata = errors.New("kafka: no specific topics to update metadata")

+ var ErrNonTransactedProducer = errors.New("transaction manager: you need to add TransactionalID to producer")

+ var ErrNotConnected = errors.New("kafka: broker not connected")

+ var ErrOutOfBrokers = errors.New("kafka: client has run out of available brokers to talk to")

+ var ErrReassignPartitions = errors.New("failed to reassign partitions for topic")

+ var ErrShuttingDown = errors.New("kafka: message received by producer in process of shutting down")

+ var ErrTransactionNotReady = errors.New("transaction manager: transaction is not ready")

+ var ErrTransitionNotAllowed = errors.New("transaction manager: invalid transition attempted")

+ var ErrTxnOffsetCommit = errors.New("transaction manager: failed to send offsets to transaction")

+ var ErrTxnUnableToParseResponse = errors.New("transaction manager: unable to parse response")

+ var ErrUnknownScramMechanism = errors.New("kafka: unknown SCRAM mechanism provided")

+ var MaxVersion = V3_6_0_0

+ var MinVersion = V0_8_2_0

+ var SupportedVersions = []KafkaVersion

+ var V0_10_0_0 = newKafkaVersion(0, 10, 0, 0)

+ var V0_10_0_1 = newKafkaVersion(0, 10, 0, 1)

+ var V0_10_1_0 = newKafkaVersion(0, 10, 1, 0)

+ var V0_10_1_1 = newKafkaVersion(0, 10, 1, 1)

+ var V0_10_2_0 = newKafkaVersion(0, 10, 2, 0)

+ var V0_10_2_1 = newKafkaVersion(0, 10, 2, 1)

+ var V0_10_2_2 = newKafkaVersion(0, 10, 2, 2)

+ var V0_11_0_0 = newKafkaVersion(0, 11, 0, 0)

+ var V0_11_0_1 = newKafkaVersion(0, 11, 0, 1)

+ var V0_11_0_2 = newKafkaVersion(0, 11, 0, 2)

+ var V0_8_2_0 = newKafkaVersion(0, 8, 2, 0)

+ var V0_8_2_1 = newKafkaVersion(0, 8, 2, 1)

+ var V0_8_2_2 = newKafkaVersion(0, 8, 2, 2)

+ var V0_9_0_0 = newKafkaVersion(0, 9, 0, 0)

+ var V0_9_0_1 = newKafkaVersion(0, 9, 0, 1)

+ var V1_0_0_0 = newKafkaVersion(1, 0, 0, 0)

+ var V1_0_1_0 = newKafkaVersion(1, 0, 1, 0)

+ var V1_0_2_0 = newKafkaVersion(1, 0, 2, 0)

+ var V1_1_0_0 = newKafkaVersion(1, 1, 0, 0)

+ var V1_1_1_0 = newKafkaVersion(1, 1, 1, 0)

+ var V2_0_0_0 = newKafkaVersion(2, 0, 0, 0)

+ var V2_0_1_0 = newKafkaVersion(2, 0, 1, 0)

+ var V2_1_0_0 = newKafkaVersion(2, 1, 0, 0)

+ var V2_1_1_0 = newKafkaVersion(2, 1, 1, 0)

+ var V2_2_0_0 = newKafkaVersion(2, 2, 0, 0)

+ var V2_2_1_0 = newKafkaVersion(2, 2, 1, 0)

+ var V2_2_2_0 = newKafkaVersion(2, 2, 2, 0)

+ var V2_3_0_0 = newKafkaVersion(2, 3, 0, 0)

+ var V2_3_1_0 = newKafkaVersion(2, 3, 1, 0)

+ var V2_4_0_0 = newKafkaVersion(2, 4, 0, 0)

+ var V2_4_1_0 = newKafkaVersion(2, 4, 1, 0)

+ var V2_5_0_0 = newKafkaVersion(2, 5, 0, 0)

+ var V2_5_1_0 = newKafkaVersion(2, 5, 1, 0)

+ var V2_6_0_0 = newKafkaVersion(2, 6, 0, 0)

+ var V2_6_1_0 = newKafkaVersion(2, 6, 1, 0)

+ var V2_6_2_0 = newKafkaVersion(2, 6, 2, 0)

+ var V2_6_3_0 = newKafkaVersion(2, 6, 3, 0)

+ var V2_7_0_0 = newKafkaVersion(2, 7, 0, 0)

+ var V2_7_1_0 = newKafkaVersion(2, 7, 1, 0)

+ var V2_7_2_0 = newKafkaVersion(2, 7, 2, 0)

+ var V2_8_0_0 = newKafkaVersion(2, 8, 0, 0)

+ var V2_8_1_0 = newKafkaVersion(2, 8, 1, 0)

+ var V2_8_2_0 = newKafkaVersion(2, 8, 2, 0)

+ var V3_0_0_0 = newKafkaVersion(3, 0, 0, 0)

+ var V3_0_1_0 = newKafkaVersion(3, 0, 1, 0)

+ var V3_0_2_0 = newKafkaVersion(3, 0, 2, 0)

+ var V3_1_0_0 = newKafkaVersion(3, 1, 0, 0)

+ var V3_1_1_0 = newKafkaVersion(3, 1, 1, 0)

+ var V3_1_2_0 = newKafkaVersion(3, 1, 2, 0)

+ var V3_2_0_0 = newKafkaVersion(3, 2, 0, 0)

+ var V3_2_1_0 = newKafkaVersion(3, 2, 1, 0)

+ var V3_2_2_0 = newKafkaVersion(3, 2, 2, 0)

+ var V3_2_3_0 = newKafkaVersion(3, 2, 3, 0)

+ var V3_3_0_0 = newKafkaVersion(3, 3, 0, 0)

+ var V3_3_1_0 = newKafkaVersion(3, 3, 1, 0)

+ var V3_3_2_0 = newKafkaVersion(3, 3, 2, 0)

+ var V3_4_0_0 = newKafkaVersion(3, 4, 0, 0)

+ var V3_4_1_0 = newKafkaVersion(3, 4, 1, 0)

+ var V3_5_0_0 = newKafkaVersion(3, 5, 0, 0)

+ var V3_5_1_0 = newKafkaVersion(3, 5, 1, 0)

+ var V3_6_0_0 = newKafkaVersion(3, 6, 0, 0)

+ func ResponseHeaderVersion(apiKey, apiVersion int16) int16

+ func VersionedDecode(buf []byte, in versionedDecoder, version int16) (int, error)

+ type AbortedTransaction struct

+ FirstOffset int64

+ ProducerID int64

+ type ByteEncoder []byte

+ func (b ByteEncoder) Encode() ([]byte, error)

+ func (b ByteEncoder) Length() int

+ type CompressionCodec int8

+ const CompressionGZIP

+ const CompressionLZ4

+ const CompressionLevelDefault

+ const CompressionNone

+ const CompressionSnappy

+ const CompressionZSTD

+ func (cc *CompressionCodec) UnmarshalText(text []byte) error

+ func (cc CompressionCodec) MarshalText() ([]byte, error)

+ func (cc CompressionCodec) String() string

+ type ConfigurationError string

+ func (err ConfigurationError) Error() string

+ type Encoder interface

+ Encode func() ([]byte, error)

+ Length func() int

+ type FetchResponse struct

+ Blocks map[string]map[int32]*FetchResponseBlock

+ ErrorCode int16

+ LogAppendTime bool

+ SessionID int32

+ ThrottleTime time.Duration

+ Timestamp time.Time

+ Version int16

+ type FetchResponseBlock struct

+ AbortedTransactions []*AbortedTransaction

+ Err KError

+ HighWaterMarkOffset int64

+ LastRecordsBatchOffset *int64

+ LastStableOffset int64

+ LogStartOffset int64

+ Partial bool

+ PreferredReadReplica int32

+ Records *Records

+ RecordsSet []*Records

+ type KError int16

+ const ErrBrokerNotAvailable

+ const ErrClusterAuthorizationFailed

+ const ErrConcurrentTransactions

+ const ErrConsumerCoordinatorNotAvailable

+ const ErrDelegationTokenAuthDisabled

+ const ErrDelegationTokenAuthorizationFailed

+ const ErrDelegationTokenExpired

+ const ErrDelegationTokenNotFound

+ const ErrDelegationTokenOwnerMismatch

+ const ErrDelegationTokenRequestNotAllowed

+ const ErrDuplicateSequenceNumber

+ const ErrElectionNotNeeded

+ const ErrEligibleLeadersNotAvailable

+ const ErrFencedInstancedId

+ const ErrFencedLeaderEpoch

+ const ErrFetchSessionIDNotFound

+ const ErrGroupAuthorizationFailed

+ const ErrGroupIDNotFound

+ const ErrGroupMaxSizeReached

+ const ErrGroupSubscribedToTopic

+ const ErrIllegalGeneration

+ const ErrIllegalSASLState

+ const ErrInconsistentGroupProtocol

+ const ErrInvalidCommitOffsetSize

+ const ErrInvalidConfig

+ const ErrInvalidFetchSessionEpoch

+ const ErrInvalidGroupId

+ const ErrInvalidMessage

+ const ErrInvalidMessageSize

+ const ErrInvalidPartitions

+ const ErrInvalidPrincipalType

+ const ErrInvalidProducerEpoch

+ const ErrInvalidProducerIDMapping

+ const ErrInvalidRecord

+ const ErrInvalidReplicaAssignment

+ const ErrInvalidReplicationFactor

+ const ErrInvalidRequest

+ const ErrInvalidRequiredAcks

+ const ErrInvalidSessionTimeout

+ const ErrInvalidTimestamp

+ const ErrInvalidTopic

+ const ErrInvalidTransactionTimeout

+ const ErrInvalidTxnState

+ const ErrKafkaStorageError

+ const ErrLeaderNotAvailable

+ const ErrListenerNotFound

+ const ErrLogDirNotFound

+ const ErrMemberIdRequired

+ const ErrMessageSetSizeTooLarge

+ const ErrMessageSizeTooLarge

+ const ErrNetworkException

+ const ErrNoError

+ const ErrNoReassignmentInProgress

+ const ErrNonEmptyGroup

+ const ErrNotController

+ const ErrNotCoordinatorForConsumer

+ const ErrNotEnoughReplicas

+ const ErrNotEnoughReplicasAfterAppend

+ const ErrNotLeaderForPartition

+ const ErrOffsetMetadataTooLarge

+ const ErrOffsetNotAvailable

+ const ErrOffsetOutOfRange

+ const ErrOffsetsLoadInProgress

+ const ErrOperationNotAttempted

+ const ErrOutOfOrderSequenceNumber

+ const ErrPolicyViolation

+ const ErrPreferredLeaderNotAvailable

+ const ErrProducerFenced

+ const ErrReassignmentInProgress

+ const ErrRebalanceInProgress

+ const ErrReplicaNotAvailable

+ const ErrRequestTimedOut

+ const ErrSASLAuthenticationFailed

+ const ErrSecurityDisabled

+ const ErrStaleBrokerEpoch

+ const ErrStaleControllerEpochCode

+ const ErrThrottlingQuotaExceeded

+ const ErrTopicAlreadyExists

+ const ErrTopicAuthorizationFailed

+ const ErrTopicDeletionDisabled

+ const ErrTransactionCoordinatorFenced

+ const ErrTransactionalIDAuthorizationFailed

+ const ErrUnknown

+ const ErrUnknownLeaderEpoch

+ const ErrUnknownMemberId

+ const ErrUnknownProducerID

+ const ErrUnknownTopicOrPartition

+ const ErrUnstableOffsetCommit

+ const ErrUnsupportedCompressionType

+ const ErrUnsupportedForMessageFormat

+ const ErrUnsupportedSASLMechanism

+ const ErrUnsupportedVersion

+ func (err KError) Error() string

+ type KafkaVersion struct

+ func ParseKafkaVersion(s string) (KafkaVersion, error)

+ func (v KafkaVersion) IsAtLeast(other KafkaVersion) bool

+ func (v KafkaVersion) String() string

+ type Message struct

+ Codec CompressionCodec

+ CompressionLevel int

+ Key []byte

+ LogAppendTime bool

+ Set *MessageSet

+ Timestamp time.Time

+ Value []byte

+ Version int8

+ type MessageBlock struct

+ Msg *Message

+ Offset int64

+ func (msb *MessageBlock) Messages() []*MessageBlock

+ type MessageSet struct

+ Messages []*MessageBlock

+ OverflowMessage bool

+ PartialTrailingMessage bool

+ type PacketDecodingError struct

+ Info string

+ func (err PacketDecodingError) Error() string

+ type PacketEncodingError struct

+ Info string

+ func (err PacketEncodingError) Error() string

+ type ProduceRequest struct

+ Records map[string]map[int32]Records

+ RequiredAcks RequiredAcks

+ Timeout int32

+ TransactionalID *string

+ Version int16

+ type ProtocolBody interface

+ type Record struct

+ Attributes int8

+ Headers []*RecordHeader

+ Key []byte

+ OffsetDelta int64

+ TimestampDelta time.Duration

+ Value []byte

+ type RecordBatch struct

+ Codec CompressionCodec

+ CompressionLevel int

+ Control bool

+ FirstOffset int64

+ FirstSequence int32

+ FirstTimestamp time.Time

+ IsTransactional bool

+ LastOffsetDelta int32

+ LogAppendTime bool

+ MaxTimestamp time.Time

+ PartialTrailingRecord bool

+ PartitionLeaderEpoch int32

+ ProducerEpoch int16

+ ProducerID int64

+ Records []*Record

+ Version int8

+ func (b *RecordBatch) LastOffset() int64

+ type RecordHeader struct

+ Key []byte

+ Value []byte

+ type Records struct

+ MsgSet *MessageSet

+ RecordBatch *RecordBatch

+ type Request struct

+ Body ProtocolBody

+ ClientID string

+ CorrelationID int32

+ func DecodeRequest(r io.Reader) (*Request, int, error)

+ type RequiredAcks int16

+ const NoResponse

+ const WaitForAll

+ const WaitForLocal

+ type ResponseHeader struct

+ CorrelationID int32

+ Length int32

+ type StringEncoder string

+ func (s StringEncoder) Encode() ([]byte, error)

+ func (s StringEncoder) Length() int

+ type Timestamp struct

+ type ZstdDecoderParams struct

+ type ZstdEncoderParams struct

+ Level int