protocol

package
v0.4.47 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Dec 13, 2023 License: MIT Imports: 21 Imported by: 462

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Marshal

func Marshal(version int16, value interface{}) ([]byte, error)

func ReadAll

func ReadAll(b Bytes) ([]byte, error)

ReadAll is similar to ioutil.ReadAll, but it takes advantage of knowing the length of b to minimize the memory footprint.

The function returns a nil slice if b is nil.

func Register

func Register(req, res Message)

Register is automatically called by sub-packages are imported to install a new pair of request/response message types.

func RegisterOverride added in v0.4.47

func RegisterOverride(req, res Message, key OverrideTypeKey)

func Unmarshal

func Unmarshal(data []byte, version int16, value interface{}) error

func WriteRequest

func WriteRequest(w io.Writer, apiVersion int16, correlationID int32, clientID string, msg Message) error

func WriteResponse

func WriteResponse(w io.Writer, apiVersion int16, correlationID int32, msg Message) error

Types

type ApiKey

type ApiKey int16
const (
	Produce                      ApiKey = 0
	Fetch                        ApiKey = 1
	ListOffsets                  ApiKey = 2
	Metadata                     ApiKey = 3
	LeaderAndIsr                 ApiKey = 4
	StopReplica                  ApiKey = 5
	UpdateMetadata               ApiKey = 6
	ControlledShutdown           ApiKey = 7
	OffsetCommit                 ApiKey = 8
	OffsetFetch                  ApiKey = 9
	FindCoordinator              ApiKey = 10
	JoinGroup                    ApiKey = 11
	Heartbeat                    ApiKey = 12
	LeaveGroup                   ApiKey = 13
	SyncGroup                    ApiKey = 14
	DescribeGroups               ApiKey = 15
	ListGroups                   ApiKey = 16
	SaslHandshake                ApiKey = 17
	ApiVersions                  ApiKey = 18
	CreateTopics                 ApiKey = 19
	DeleteTopics                 ApiKey = 20
	DeleteRecords                ApiKey = 21
	InitProducerId               ApiKey = 22
	OffsetForLeaderEpoch         ApiKey = 23
	AddPartitionsToTxn           ApiKey = 24
	AddOffsetsToTxn              ApiKey = 25
	EndTxn                       ApiKey = 26
	WriteTxnMarkers              ApiKey = 27
	TxnOffsetCommit              ApiKey = 28
	DescribeAcls                 ApiKey = 29
	CreateAcls                   ApiKey = 30
	DeleteAcls                   ApiKey = 31
	DescribeConfigs              ApiKey = 32
	AlterConfigs                 ApiKey = 33
	AlterReplicaLogDirs          ApiKey = 34
	DescribeLogDirs              ApiKey = 35
	SaslAuthenticate             ApiKey = 36
	CreatePartitions             ApiKey = 37
	CreateDelegationToken        ApiKey = 38
	RenewDelegationToken         ApiKey = 39
	ExpireDelegationToken        ApiKey = 40
	DescribeDelegationToken      ApiKey = 41
	DeleteGroups                 ApiKey = 42
	ElectLeaders                 ApiKey = 43
	IncrementalAlterConfigs      ApiKey = 44
	AlterPartitionReassignments  ApiKey = 45
	ListPartitionReassignments   ApiKey = 46
	OffsetDelete                 ApiKey = 47
	DescribeClientQuotas         ApiKey = 48
	AlterClientQuotas            ApiKey = 49
	DescribeUserScramCredentials ApiKey = 50
	AlterUserScramCredentials    ApiKey = 51
)

func (ApiKey) MaxVersion

func (k ApiKey) MaxVersion() int16

func (ApiKey) MinVersion

func (k ApiKey) MinVersion() int16

func (ApiKey) SelectVersion

func (k ApiKey) SelectVersion(minVersion, maxVersion int16) int16

func (ApiKey) String

func (k ApiKey) String() string

type Attributes

type Attributes int16

Attributes is a bitset representing special attributes set on records.

const (
	Gzip          Attributes = Attributes(compress.Gzip)   // 1
	Snappy        Attributes = Attributes(compress.Snappy) // 2
	Lz4           Attributes = Attributes(compress.Lz4)    // 3
	Zstd          Attributes = Attributes(compress.Zstd)   // 4
	Transactional Attributes = 1 << 4
	Control       Attributes = 1 << 5
)

func (Attributes) Compression

func (a Attributes) Compression() compress.Compression

func (Attributes) Control

func (a Attributes) Control() bool

func (Attributes) String

func (a Attributes) String() string

func (Attributes) Transactional

func (a Attributes) Transactional() bool

type Broker

type Broker struct {
	Rack string
	Host string
	Port int32
	ID   int32
}

func (Broker) Format

func (b Broker) Format(w fmt.State, v rune)

func (Broker) String

func (b Broker) String() string

type BrokerMessage

type BrokerMessage interface {
	// Given a representation of the kafka cluster state as argument, returns
	// the broker that the message should be routed to.
	Broker(Cluster) (Broker, error)
}

BrokerMessage is an extension of the Message interface implemented by some request types to customize the broker assignment logic.

type Bytes

type Bytes interface {
	io.ReadCloser
	// Returns the number of bytes remaining to be read from the payload.
	Len() int
}

Bytes is an interface implemented by types that represent immutable sequences of bytes.

Bytes values are used to abstract the location where record keys and values are read from (e.g. in-memory buffers, network sockets, files).

The Close method should be called to release resources held by the object when the program is done with it.

Bytes values are generally not safe to use concurrently from multiple goroutines.

func NewBytes

func NewBytes(b []byte) Bytes

NewBytes constructs a Bytes value from b.

The returned value references b, it does not make a copy of the backing array.

If b is nil, nil is returned to represent a null BYTES value in the kafka protocol.

type Cluster

type Cluster struct {
	ClusterID  string
	Controller int32
	Brokers    map[int32]Broker
	Topics     map[string]Topic
}

func (Cluster) BrokerIDs

func (c Cluster) BrokerIDs() []int32

func (Cluster) Format

func (c Cluster) Format(w fmt.State, _ rune)

func (Cluster) IsZero

func (c Cluster) IsZero() bool

func (Cluster) TopicNames

func (c Cluster) TopicNames() []string

type Conn

type Conn struct {
	// contains filtered or unexported fields
}

func NewConn

func NewConn(conn net.Conn, clientID string) *Conn

func (*Conn) Close

func (c *Conn) Close() error

func (*Conn) Discard

func (c *Conn) Discard(n int) (int, error)

func (*Conn) LocalAddr

func (c *Conn) LocalAddr() net.Addr

func (*Conn) Peek

func (c *Conn) Peek(n int) ([]byte, error)

func (*Conn) Read

func (c *Conn) Read(b []byte) (int, error)

func (*Conn) RemoteAddr

func (c *Conn) RemoteAddr() net.Addr

func (*Conn) RoundTrip

func (c *Conn) RoundTrip(msg Message) (Message, error)

func (*Conn) SetDeadline

func (c *Conn) SetDeadline(t time.Time) error

func (*Conn) SetReadDeadline

func (c *Conn) SetReadDeadline(t time.Time) error

func (*Conn) SetVersions

func (c *Conn) SetVersions(versions map[ApiKey]int16)

func (*Conn) SetWriteDeadline

func (c *Conn) SetWriteDeadline(t time.Time) error

func (*Conn) String

func (c *Conn) String() string

func (*Conn) Write

func (c *Conn) Write(b []byte) (int, error)

type ControlBatch

type ControlBatch struct {
	Attributes           Attributes
	PartitionLeaderEpoch int32
	BaseOffset           int64
	ProducerID           int64
	ProducerEpoch        int16
	BaseSequence         int32
	Records              RecordReader
}

ControlBatch is an implementation of the RecordReader interface representing control batches returned by kafka brokers.

func NewControlBatch

func NewControlBatch(records ...ControlRecord) *ControlBatch

NewControlBatch constructs a control batch from the list of records passed as arguments.

func (*ControlBatch) Offset

func (c *ControlBatch) Offset() int64

func (*ControlBatch) ReadControlRecord

func (c *ControlBatch) ReadControlRecord() (*ControlRecord, error)

func (*ControlBatch) ReadRecord

func (c *ControlBatch) ReadRecord() (*Record, error)

func (*ControlBatch) Version

func (c *ControlBatch) Version() int

type ControlRecord

type ControlRecord struct {
	Offset  int64
	Time    time.Time
	Version int16
	Type    int16
	Data    []byte
	Headers []Header
}

ControlRecord represents a record read from a control batch.

func ReadControlRecord

func ReadControlRecord(r *Record) (*ControlRecord, error)

func (*ControlRecord) Key

func (cr *ControlRecord) Key() Bytes

func (*ControlRecord) Record

func (cr *ControlRecord) Record() Record

func (*ControlRecord) Value

func (cr *ControlRecord) Value() Bytes

type Error

type Error string

Error represents client-side protocol errors.

const (
	// ErrNoTopic is returned when a request needs to be sent to a specific.
	ErrNoTopic Error = "topic not found"

	// ErrNoPartition is returned when a request needs to be sent to a specific
	// partition, but the client did not find it in the cluster metadata.
	ErrNoPartition Error = "topic partition not found"

	// ErrNoLeader is returned when a request needs to be sent to a partition
	// leader, but the client could not determine what the leader was at this
	// time.
	ErrNoLeader Error = "topic partition has no leader"

	// ErrNoRecord is returned when attempting to write a message containing an
	// empty record set (which kafka forbids).
	//
	// We handle this case client-side because kafka will close the connection
	// that it received an empty produce request on, causing all concurrent
	// requests to be aborted.
	ErrNoRecord Error = "record set contains no records"

	// ErrNoReset is returned by ResetRecordReader when the record reader does
	// not support being reset.
	ErrNoReset Error = "record sequence does not support reset"
)

func Errorf

func Errorf(msg string, args ...interface{}) Error

func (Error) Error

func (e Error) Error() string

type GroupMessage

type GroupMessage interface {
	// Returns the group configured on the message.
	Group() string
}

GroupMessage is an extension of the Message interface implemented by some request types to inform the program that they should be routed to a group coordinator.

type Header struct {
	Key   string
	Value []byte
}

Header represents a single entry in a list of record headers.

type Merger

type Merger interface {
	// Given a list of message and associated results, merge them back into a
	// response (or an error). The results must be either Message or error
	// values, other types should trigger a panic.
	Merge(messages []Message, results []interface{}) (Message, error)
}

Merger is an interface implemented by messages which can merge multiple results into one response.

type Message

type Message interface {
	ApiKey() ApiKey
}

Message is an interface implemented by all request and response types of the kafka protocol.

This interface is used mostly as a safe-guard to provide a compile-time check for values passed to functions dealing kafka message types.

func ReadRequest

func ReadRequest(r io.Reader) (apiVersion int16, correlationID int32, clientID string, msg Message, err error)

func ReadResponse

func ReadResponse(r io.Reader, apiKey ApiKey, apiVersion int16) (correlationID int32, msg Message, err error)

func Result

func Result(r interface{}) (Message, error)

Result converts r to a Message or an error, or panics if r could not be converted to these types.

func RoundTrip

func RoundTrip(rw io.ReadWriter, apiVersion int16, correlationID int32, clientID string, req Message) (Message, error)

RoundTrip sends a request to a kafka broker and returns the response.

type MessageSet

type MessageSet struct {
	Attributes Attributes
	BaseOffset int64
	Records    RecordReader
}

MessageSet is an implementation of the RecordReader interface representing regular message sets (v1).

func (*MessageSet) Offset

func (m *MessageSet) Offset() int64

func (*MessageSet) ReadRecord

func (m *MessageSet) ReadRecord() (*Record, error)

func (*MessageSet) Version

func (m *MessageSet) Version() int

type OverrideTypeKey added in v0.4.47

type OverrideTypeKey int16
const (
	RawProduceOverride OverrideTypeKey = 0
)

type OverrideTypeMessage added in v0.4.47

type OverrideTypeMessage interface {
	TypeKey() OverrideTypeKey
}

OverrideTypeMessage is an interface implemented by messages that want to override the standard request/response types for a given API.

type Partition

type Partition struct {
	ID       int32
	Error    int16
	Leader   int32
	Replicas []int32
	ISR      []int32
	Offline  []int32
}

type PreparedMessage

type PreparedMessage interface {
	// Prepares the message before being sent to a kafka broker using the API
	// version passed as argument.
	Prepare(apiVersion int16)
}

PreparedMessage is an extension of the Message interface implemented by some request types which may need to run some pre-processing on their state before being sent.

type RawExchanger added in v0.4.31

type RawExchanger interface {
	// Required should return true when a RawExchange is needed.
	// The passed in versions are the negotiated versions for the connection
	// performing the request.
	Required(versions map[ApiKey]int16) bool
	// RawExchange is given the raw connection to the broker and the Message
	// is responsible for writing itself to the connection as well as reading
	// the response.
	RawExchange(rw io.ReadWriter) (Message, error)
}

RawExchanger is an extention to the Message interface to allow messages to control the request response cycle for the message. This is currently only used to facilitate v0 SASL Authenticate requests being written in a non-standard fashion when the SASL Handshake was done at v0 but not when done at v1.

type RawRecordSet added in v0.4.47

type RawRecordSet struct {
	// Reader exposes the raw sequence of record set bytes.
	Reader io.Reader
}

RawRecordSet represents a record set for a RawProduce request. The record set is represented as a raw sequence of pre-encoded record set bytes.

func (*RawRecordSet) ReadFrom added in v0.4.47

func (rrs *RawRecordSet) ReadFrom(r io.Reader) (int64, error)

ReadFrom reads the representation of a record set from r into rrs. It re-uses the existing RecordSet.ReadFrom implementation to first read/decode data into a RecordSet, then writes/encodes the RecordSet to a buffer referenced by the RawRecordSet.

Note: re-using the RecordSet.ReadFrom implementation makes this suboptimal from a performance standpoint as it require an extra copy of the record bytes. Holding off on optimizing, as this code path is only invoked in tests.

func (*RawRecordSet) WriteTo added in v0.4.47

func (rrs *RawRecordSet) WriteTo(w io.Writer) (int64, error)

WriteTo writes the RawRecordSet to an io.Writer. Since this is a raw record set representation, all that is done here is copying bytes from the underlying reader to the specified writer.

type Record

type Record struct {
	// The offset at which the record exists in a topic partition. This value
	// is ignored in produce requests.
	Offset int64

	// Returns the time of the record. This value may be omitted in produce
	// requests to let kafka set the time when it saves the record.
	Time time.Time

	// Returns a byte sequence containing the key of this record. The returned
	// sequence may be nil to indicate that the record has no key. If the record
	// is part of a RecordSet, the content of the key must remain valid at least
	// until the record set is closed (or until the key is closed).
	Key Bytes

	// Returns a byte sequence containing the value of this record. The returned
	// sequence may be nil to indicate that the record has no value. If the
	// record is part of a RecordSet, the content of the value must remain valid
	// at least until the record set is closed (or until the value is closed).
	Value Bytes

	// Returns the list of headers associated with this record. The returned
	// slice may be reused across calls, the program should use it as an
	// immutable value.
	Headers []Header
}

Record is an interface representing a single kafka record.

Record values are not safe to use concurrently from multiple goroutines.

type RecordBatch

type RecordBatch struct {
	Attributes           Attributes
	PartitionLeaderEpoch int32
	BaseOffset           int64
	ProducerID           int64
	ProducerEpoch        int16
	BaseSequence         int32
	Records              RecordReader
}

RecordBatch is an implementation of the RecordReader interface representing regular record batches (v2).

func (*RecordBatch) Offset

func (r *RecordBatch) Offset() int64

func (*RecordBatch) ReadRecord

func (r *RecordBatch) ReadRecord() (*Record, error)

func (*RecordBatch) Version

func (r *RecordBatch) Version() int

type RecordReader

type RecordReader interface {
	// Returns the next record in the set, or io.EOF if the end of the sequence
	// has been reached.
	//
	// The returned Record is guaranteed to be valid until the next call to
	// ReadRecord. If the program needs to retain the Record value it must make
	// a copy.
	ReadRecord() (*Record, error)
}

RecordReader is an interface representing a sequence of records. Record sets are used in both produce and fetch requests to represent the sequence of records that are sent to or receive from kafka brokers.

RecordSet values are not safe to use concurrently from multiple goroutines.

func MultiRecordReader

func MultiRecordReader(batches ...RecordReader) RecordReader

MultiRecordReader merges multiple record batches into one.

func NewRecordReader

func NewRecordReader(records ...Record) RecordReader

NewRecordReader constructs a reader exposing the records passed as arguments.

type RecordSet

type RecordSet struct {
	// The message version that this record set will be represented as, valid
	// values are 1, or 2.
	//
	// When reading, this is the value of the highest version used in the
	// batches that compose the record set.
	//
	// When writing, this value dictates the format that the records will be
	// encoded in.
	Version int8

	// Attributes set on the record set.
	//
	// When reading, the attributes are the combination of all attributes in
	// the batches that compose the record set.
	//
	// When writing, the attributes apply to the whole sequence of records in
	// the set.
	Attributes Attributes

	// A reader exposing the sequence of records.
	//
	// When reading a RecordSet from an io.Reader, the Records field will be a
	// *RecordStream. If the program needs to access the details of each batch
	// that compose the stream, it may use type assertions to access the
	// underlying types of each batch.
	Records RecordReader
}

RecordSet represents a sequence of records in Produce requests and Fetch responses. All v0, v1, and v2 formats are supported.

func (*RecordSet) ReadFrom

func (rs *RecordSet) ReadFrom(r io.Reader) (int64, error)

ReadFrom reads the representation of a record set from r into rs, returning the number of bytes consumed from r, and an non-nil error if the record set could not be read.

func (*RecordSet) WriteTo

func (rs *RecordSet) WriteTo(w io.Writer) (int64, error)

WriteTo writes the representation of rs into w. The value of rs.Version dictates which format that the record set will be represented as.

The error will be ErrNoRecord if rs contained no records.

Note: since this package is only compatible with kafka 0.10 and above, the method never produces messages in version 0. If rs.Version is zero, the method defaults to producing messages in version 1.

type RecordStream

type RecordStream struct {
	Records []RecordReader
	// contains filtered or unexported fields
}

RecordStream is an implementation of the RecordReader interface which combines multiple underlying RecordReader and only expose records that are not from control batches.

func (*RecordStream) ReadRecord

func (s *RecordStream) ReadRecord() (*Record, error)

type Splitter

type Splitter interface {
	// For a given cluster layout, returns the list of messages constructed
	// from the receiver for each requests that should be sent to the cluster.
	// The second return value is a Merger which can be used to merge back the
	// results of each request into a single message (or an error).
	Split(Cluster) ([]Message, Merger, error)
}

Splitter is an interface implemented by messages that can be split into multiple requests and have their results merged back by a Merger.

type Topic

type Topic struct {
	Name       string
	Error      int16
	Partitions map[int32]Partition
}

type TopicError

type TopicError struct {
	Topic string
	Err   error
}

func NewErrNoTopic

func NewErrNoTopic(topic string) *TopicError

func NewTopicError

func NewTopicError(topic string, err error) *TopicError

func (*TopicError) Error

func (e *TopicError) Error() string

func (*TopicError) Unwrap

func (e *TopicError) Unwrap() error

type TopicPartitionError

type TopicPartitionError struct {
	Topic     string
	Partition int32
	Err       error
}

func NewErrNoLeader

func NewErrNoLeader(topic string, partition int32) *TopicPartitionError

func NewErrNoPartition

func NewErrNoPartition(topic string, partition int32) *TopicPartitionError

func NewTopicPartitionError

func NewTopicPartitionError(topic string, partition int32, err error) *TopicPartitionError

func (*TopicPartitionError) Error

func (e *TopicPartitionError) Error() string

func (*TopicPartitionError) Unwrap

func (e *TopicPartitionError) Unwrap() error

type TransactionalMessage added in v0.4.20

type TransactionalMessage interface {
	// Returns the transactional id configured on the message.
	Transaction() string
}

TransactionalMessage is an extension of the Message interface implemented by some request types to inform the program that they should be routed to a transaction coordinator.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL