sarama

type Broker ¶

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

Broker represents a single Kafka broker connection. All operations on this object are entirely concurrency-safe.

broker := NewBroker("localhost:9092")
err := broker.Open(nil)
if err != nil {
	return err
}
defer broker.Close()

request := MetadataRequest{Topics: []string{"myTopic"}}
response, err := broker.GetMetadata("myClient", &request)
if err != nil {
	return err
}

fmt.Println("There are", len(response.Topics), "topics active in the cluster.")

return nil

Output:

func NewBroker(addr string) *Broker

func (b *Broker) Addr() string

func (b *Broker) Close() (err error)

func (b *Broker) CommitOffset(clientID string, request *OffsetCommitRequest) (*OffsetCommitResponse, error)

func (b *Broker) Connected() (bool, error)

func (b *Broker) Fetch(clientID string, request *FetchRequest) (*FetchResponse, error)

func (b *Broker) FetchOffset(clientID string, request *OffsetFetchRequest) (*OffsetFetchResponse, error)

func (b *Broker) GetAvailableOffsets(clientID string, request *OffsetRequest) (*OffsetResponse, error)

func (b *Broker) GetConsumerMetadata(clientID string, request *ConsumerMetadataRequest) (*ConsumerMetadataResponse, error)

func (b *Broker) GetMetadata(clientID string, request *MetadataRequest) (*MetadataResponse, error)

func (b *Broker) ID() int32

func (b *Broker) Open(conf *BrokerConfig) error

func (b *Broker) Produce(clientID string, request *ProduceRequest) (*ProduceResponse, error)

type BrokerConfig ¶

type BrokerConfig struct {
	MaxOpenRequests int           // How many outstanding requests the broker is allowed to have before blocking attempts to send.
	DialTimeout     time.Duration // How long to wait for the initial connection to succeed before timing out and returning an error.
	ReadTimeout     time.Duration // How long to wait for a response before timing out and returning an error.
	WriteTimeout    time.Duration // How long to wait for a transmit to succeed before timing out and returning an error.
}

BrokerConfig is used to pass multiple configuration options to Broker.Open.

func NewBrokerConfig() *BrokerConfig

func (config *BrokerConfig) Validate() error

type ByteEncoder ¶

type ByteEncoder []byte

ByteEncoder implements the Encoder interface for Go byte slices so that you can do things like

producer.SendMessage(nil, sarama.ByteEncoder([]byte{0x00}))

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

func (b ByteEncoder) Length() int

type Client ¶

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

Client is a generic Kafka client. It manages connections to one or more Kafka brokers. You MUST call Close() on a client to avoid leaks, it will not be garbage-collected automatically when it passes out of scope. A single client can be safely shared by multiple concurrent Producers and Consumers.

func NewClient(id string, addrs []string, config *ClientConfig) (*Client, error)

func (client *Client) Close() error

func (client *Client) Closed() bool

func (client *Client) GetOffset(topic string, partitionID int32, where OffsetTime) (int64, error)

func (client *Client) Leader(topic string, partitionID int32) (*Broker, error)

func (client *Client) Partitions(topic string) ([]int32, error)

func (client *Client) RefreshAllMetadata() error

func (client *Client) RefreshTopicMetadata(topics ...string) error

func (client *Client) Topics() ([]string, error)

type ClientConfig ¶

type ClientConfig struct {
	MetadataRetries            int           // How many times to retry a metadata request when a partition is in the middle of leader election.
	WaitForElection            time.Duration // How long to wait for leader election to finish between retries.
	DefaultBrokerConf          *BrokerConfig // Default configuration for broker connections created by this client.
	BackgroundRefreshFrequency time.Duration // How frequently the client will refresh the cluster metadata in the background. Defaults to 10 minutes. Set to 0 to disable.
}

ClientConfig is used to pass multiple configuration options to NewClient.

func NewClientConfig() *ClientConfig

func (config *ClientConfig) Validate() error

type CompressionCodec ¶

type CompressionCodec int8

CompressionCodec represents the various compression codecs recognized by Kafka in messages.

const (
	CompressionNone   CompressionCodec = 0
	CompressionGZIP   CompressionCodec = 1
	CompressionSnappy CompressionCodec = 2
)

type ConfigurationError ¶

type ConfigurationError string

ConfigurationError is the type of error returned from NewClient, NewProducer or NewConsumer when the specified configuration is invalid.

func (err ConfigurationError) Error() string

type ConstantPartitioner ¶

type ConstantPartitioner struct {
	Constant int32
}

ConstantPartitioner implements the Partitioner interface by just returning a constant value.

func (p *ConstantPartitioner) Partition(key Encoder, numPartitions int32) int32

type Consumer ¶

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

Consumer processes Kafka messages from a given topic and partition. You MUST call Close() on a consumer to avoid leaks, it will not be garbage-collected automatically when it passes out of scope (this is in addition to calling Close on the underlying client, which is still necessary).

client, err := NewClient("my_client", []string{"localhost:9092"}, nil)
if err != nil {
	panic(err)
} else {
	fmt.Println("> connected")
}
defer client.Close()

consumer, err := NewConsumer(client, "my_topic", 0, "my_consumer_group", NewConsumerConfig())
if err != nil {
	panic(err)
} else {
	fmt.Println("> consumer ready")
}
defer consumer.Close()

msgCount := 0
consumerLoop:
for {
	select {
	case event := <-consumer.Events():
		if event.Err != nil {
			panic(event.Err)
		}
		msgCount++
	case <-time.After(5 * time.Second):
		fmt.Println("> timed out")
		break consumerLoop
	}
}
fmt.Println("Got", msgCount, "messages.")

Output:

func NewConsumer(client *Client, topic string, partition int32, group string, config *ConsumerConfig) (*Consumer, error)

func (c *Consumer) Close() error

func (c *Consumer) Events() <-chan *ConsumerEvent

type ConsumerConfig ¶

type ConsumerConfig struct {
	// The default (maximum) amount of data to fetch from the broker in each request. The default is 32768 bytes.
	DefaultFetchSize int32
	// The minimum amount of data to fetch in a request - the broker will wait until at least this many bytes are available.
	// The default is 1, as 0 causes the consumer to spin when no messages are available.
	MinFetchSize int32
	// The maximum permittable message size - messages larger than this will return MessageTooLarge. The default of 0 is
	// treated as no limit.
	MaxMessageSize int32
	// The maximum amount of time the broker will wait for MinFetchSize bytes to become available before it
	// returns fewer than that anyways. The default is 250ms, since 0 causes the consumer to spin when no events are available.
	// 100-500ms is a reasonable range for most cases. Kafka only supports precision up to milliseconds; nanoseconds will be truncated.
	MaxWaitTime time.Duration

	// The method used to determine at which offset to begin consuming messages.
	OffsetMethod OffsetMethod
	// Interpreted differently according to the value of OffsetMethod.
	OffsetValue int64

	// The number of events to buffer in the Events channel. Having this non-zero permits the
	// consumer to continue fetching messages in the background while client code consumes events,
	// greatly improving throughput. The default is 16.
	EventBufferSize int
}

ConsumerConfig is used to pass multiple configuration options to NewConsumer.

func NewConsumerConfig() *ConsumerConfig

func (config *ConsumerConfig) Validate() error

type ConsumerEvent ¶

type ConsumerEvent struct {
	Key, Value []byte
	Topic      string
	Partition  int32
	Offset     int64
	Err        error
}

ConsumerEvent is what is provided to the user when an event occurs. It is either an error (in which case Err is non-nil) or a message (in which case Err is nil and Offset, Key, and Value are set). Topic and Partition are always set.

type ConsumerMetadataRequest ¶

type ConsumerMetadataRequest struct {
	ConsumerGroup string
}

type ConsumerMetadataResponse ¶

type ConsumerMetadataResponse struct {
	Err             KError
	CoordinatorId   int32
	CoordinatorHost string
	CoordinatorPort int32
}

type DecodingError ¶

type DecodingError struct {
	Info string
}

DecodingError is returned when there was an error (other than truncated data) decoding the Kafka broker's response. This can be a bad CRC or length field, or any other invalid value.

func (err DecodingError) Error() string

type DroppedMessagesError ¶

type DroppedMessagesError struct {
	DroppedMessages int
	Err             error
}

DroppedMessagesError is returned from a producer when messages weren't able to be successfully delivered to a broker.

func (err DroppedMessagesError) Error() string

type Encoder ¶

type Encoder interface {
	Encode() ([]byte, error)
	Length() int
}

Encoder is a simple interface for any type that can be encoded as an array of bytes in order to be sent as the key or value of a Kafka message. Length() is provided as an optimization, and must return the same as len() on the result of Encode().

type FetchRequest ¶

type FetchRequest struct {
	MaxWaitTime int32
	MinBytes    int32
	// contains filtered or unexported fields
}

func (f *FetchRequest) AddBlock(topic string, partitionID int32, fetchOffset int64, maxBytes int32)

type FetchResponse ¶

type FetchResponse struct {
	Blocks map[string]map[int32]*FetchResponseBlock
}

func (fr *FetchResponse) AddMessage(topic string, partition int32, key, value Encoder, offset int64)

func (fr *FetchResponse) GetBlock(topic string, partition int32) *FetchResponseBlock

type FetchResponseBlock ¶

type FetchResponseBlock struct {
	Err                 KError
	HighWaterMarkOffset int64
	MsgSet              MessageSet
}

type HashPartitioner ¶

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

HashPartitioner implements the Partitioner interface. If the key is nil, or fails to encode, then a random partition is chosen. Otherwise the FNV-1a hash of the encoded bytes is used modulus the number of partitions. This ensures that messages with the same key always end up on the same partition.

func (p *HashPartitioner) Partition(key Encoder, numPartitions int32) int32

type KError ¶

type KError int16

KError is the type of error that can be returned directly by the Kafka broker. See https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ErrorCodes

const (
	NoError                         KError = 0
	Unknown                         KError = -1
	OffsetOutOfRange                KError = 1
	InvalidMessage                  KError = 2
	UnknownTopicOrPartition         KError = 3
	InvalidMessageSize              KError = 4
	LeaderNotAvailable              KError = 5
	NotLeaderForPartition           KError = 6
	RequestTimedOut                 KError = 7
	BrokerNotAvailable              KError = 8
	ReplicaNotAvailable             KError = 9
	MessageSizeTooLarge             KError = 10
	StaleControllerEpochCode        KError = 11
	OffsetMetadataTooLarge          KError = 12
	OffsetsLoadInProgress           KError = 14
	ConsumerCoordinatorNotAvailable KError = 15
	NotCoordinatorForConsumer       KError = 16
)

func (err KError) Error() string

type Message ¶

type Message struct {
	Codec CompressionCodec // codec used to compress the message contents
	Key   []byte           // the message key, may be nil
	Value []byte           // the message contents
	Set   *MessageSet      // the message set a message might wrap
	// contains filtered or unexported fields
}

type MessageBlock ¶

type MessageBlock struct {
	Offset int64
	Msg    *Message
}

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

type MessageSet ¶

type MessageSet struct {
	PartialTrailingMessage bool // whether the set on the wire contained an incomplete trailing MessageBlock
	Messages               []*MessageBlock
}

type MessageToSend ¶

type MessageToSend struct {
	Topic      string
	Key, Value Encoder
	// contains filtered or unexported fields
}

MessageToSend is the collection of elements passed to the Producer in order to send a message.

func (m *MessageToSend) Offset() int64

func (m *MessageToSend) Partition() int32

type MetadataRequest ¶

type MetadataRequest struct {
	Topics []string
}

type MetadataResponse ¶

type MetadataResponse struct {
	Brokers []*Broker
	Topics  []*TopicMetadata
}

func (m *MetadataResponse) AddBroker(addr string, id int32)

func (m *MetadataResponse) AddTopicPartition(topic string, partition, brokerID int32)

type MockBroker ¶

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

MockBroker is a mock Kafka broker. It consists of a TCP server on a kernel-selected localhost port that accepts a single connection. It reads Kafka requests from that connection and returns each response from the channel provided at creation-time (if a response has a len of 0, nothing is sent, if a response the server sleeps for 250ms instead of reading a request).

When running tests with one of these, it is strongly recommended to specify a timeout to `go test` so that if the broker hangs waiting for a response, the test panics.

It is not necessary to prefix message length or correlation ID to your response bytes, the server does that automatically as a convenience.

func NewMockBroker(t TestState, brokerID int32) *MockBroker

func (b *MockBroker) Addr() string

func (b *MockBroker) BrokerID() int32

func (b *MockBroker) Close()

func (b *MockBroker) Port() int32

func (b *MockBroker) Returns(e encoder)

type OffsetCommitRequest ¶

type OffsetCommitRequest struct {
	ConsumerGroup string
	// contains filtered or unexported fields
}

func (r *OffsetCommitRequest) AddBlock(topic string, partitionID int32, offset int64, timestamp int64, metadata string)

type OffsetCommitResponse ¶

type OffsetCommitResponse struct {
	Errors map[string]map[int32]KError
}

type OffsetFetchRequest ¶

type OffsetFetchRequest struct {
	ConsumerGroup string
	// contains filtered or unexported fields
}

func (r *OffsetFetchRequest) AddPartition(topic string, partitionID int32)

type OffsetFetchResponse ¶

type OffsetFetchResponse struct {
	Blocks map[string]map[int32]*OffsetFetchResponseBlock
}

type OffsetFetchResponseBlock ¶

type OffsetFetchResponseBlock struct {
	Offset   int64
	Metadata string
	Err      KError
}

type OffsetMethod ¶

type OffsetMethod int

OffsetMethod is passed in ConsumerConfig to tell the consumer how to determine the starting offset.

const (
	// OffsetMethodManual causes the consumer to interpret the OffsetValue in the ConsumerConfig as the
	// offset at which to start, allowing the user to manually specify their desired starting offset.
	OffsetMethodManual OffsetMethod = iota
	// OffsetMethodNewest causes the consumer to start at the most recent available offset, as
	// determined by querying the broker.
	OffsetMethodNewest
	// OffsetMethodOldest causes the consumer to start at the oldest available offset, as
	// determined by querying the broker.
	OffsetMethodOldest
)

type OffsetRequest ¶

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

func (r *OffsetRequest) AddBlock(topic string, partitionID int32, time OffsetTime, maxOffsets int32)

type OffsetResponse ¶

type OffsetResponse struct {
	Blocks map[string]map[int32]*OffsetResponseBlock
}

func (r *OffsetResponse) AddTopicPartition(topic string, partition int32, offset int64)

func (r *OffsetResponse) GetBlock(topic string, partition int32) *OffsetResponseBlock

type OffsetResponseBlock ¶

type OffsetResponseBlock struct {
	Err     KError
	Offsets []int64
}

type OffsetTime ¶

type OffsetTime int64

OffsetTime is used in Offset Requests to ask for all messages before a certain time. Any positive int64 value will be interpreted as milliseconds, or use the special constants defined here.

const (
	// LatestOffsets askes for the latest offsets.
	LatestOffsets OffsetTime = -1
	// EarliestOffset askes for the earliest available offset. Note that because offsets are pulled in descending order,
	// asking for the earliest offset will always return you a single element.
	EarliestOffset OffsetTime = -2
)

type PartitionMetadata ¶

type PartitionMetadata struct {
	Err      KError
	ID       int32
	Leader   int32
	Replicas []int32
	Isr      []int32
}

type Partitioner ¶

type Partitioner interface {
	Partition(key Encoder, numPartitions int32) int32
}

Partitioner is anything that, given a Kafka message key and a number of partitions indexed [0...numPartitions-1], decides to which partition to send the message. RandomPartitioner, RoundRobinPartitioner and HashPartitioner are provided as simple default implementations.

func NewHashPartitioner() Partitioner

func NewRandomPartitioner() Partitioner

func NewRoundRobinPartitioner() Partitioner

type PartitionerConstructor ¶

type PartitionerConstructor func() Partitioner

PartitionerConstructor is the type for a function capable of constructing new Partitioners.

type ProduceError ¶

type ProduceError struct {
	Msg *MessageToSend
	Err error
}

ProduceError is the type of error generated when the producer fails to deliver a message. It contains the original MessageToSend as well as the actual error value. If the AckSuccesses configuration value is set to true then every message sent generates a ProduceError, but successes will have a nil Err field.

type ProduceErrors ¶

type ProduceErrors []*ProduceError

ProduceErrors is a type that wraps a batch of "ProduceError"s and implements the Error interface. It can be returned from the Producer's Close method to avoid the need to manually drain the Errors channel when closing a producer.

func (pe ProduceErrors) Error() string

type ProduceRequest ¶

type ProduceRequest struct {
	RequiredAcks RequiredAcks
	Timeout      int32
	// contains filtered or unexported fields
}

func (p *ProduceRequest) AddMessage(topic string, partition int32, msg *Message)

func (p *ProduceRequest) AddSet(topic string, partition int32, set *MessageSet)

type ProduceResponse ¶

type ProduceResponse struct {
	Blocks map[string]map[int32]*ProduceResponseBlock
}

func (pr *ProduceResponse) AddTopicPartition(topic string, partition int32, err KError)

func (pr *ProduceResponse) GetBlock(topic string, partition int32) *ProduceResponseBlock

type ProduceResponseBlock ¶

type ProduceResponseBlock struct {
	Err    KError
	Offset int64
}

type Producer ¶

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

Producer publishes Kafka messages. It routes messages to the correct broker for the provided topic-partition, refreshing metadata as appropriate, and parses responses for errors. You must read from the Errors() channel or the producer will deadlock. You must call Close() on a producer to avoid leaks: it will not be garbage-collected automatically when it passes out of scope (this is in addition to calling Close on the underlying client, which is still necessary).

client, err := NewClient("client_id", []string{"localhost:9092"}, NewClientConfig())
if err != nil {
	panic(err)
} else {
	fmt.Println("> connected")
}
defer client.Close()

producer, err := NewProducer(client, nil)
if err != nil {
	panic(err)
}
defer producer.Close()

for {
	select {
	case producer.Input() <- &MessageToSend{Topic: "my_topic", Key: nil, Value: StringEncoder("testing 123")}:
		fmt.Println("> message queued")
	case err := <-producer.Errors():
		panic(err.Err)
	}
}

Output:

func NewProducer(client *Client, config *ProducerConfig) (*Producer, error)

func (p *Producer) Close() error

func (p *Producer) Errors() <-chan *ProduceError

func (p *Producer) Input() chan<- *MessageToSend

func (p *Producer) Successes() <-chan *MessageToSend

type ProducerConfig ¶

type ProducerConfig struct {
	Partitioner       PartitionerConstructor // Generates partitioners for choosing the partition to send messages to (defaults to hash).
	RequiredAcks      RequiredAcks           // The level of acknowledgement reliability needed from the broker (defaults to WaitForLocal).
	Timeout           time.Duration          // The maximum duration the broker will wait the receipt of the number of RequiredAcks. This is only relevant when RequiredAcks is set to WaitForAll or a number > 1. Only supports millisecond resolution, nanoseconds will be truncated.
	Compression       CompressionCodec       // The type of compression to use on messages (defaults to no compression).
	FlushMsgCount     int                    // The number of messages needed to trigger a flush.
	FlushFrequency    time.Duration          // If this amount of time elapses without a flush, one will be queued.
	FlushByteCount    int                    // If this many bytes of messages are accumulated, a flush will be triggered.
	AckSuccesses      bool                   // If enabled, successfully delivered messages will be returned on the Successes channel.
	MaxMessageBytes   int                    // The maximum permitted size of a message (defaults to 1000000)
	ChannelBufferSize int                    // The size of the buffers of the channels between the different goroutines. Defaults to 0 (unbuffered).
}

ProducerConfig is used to pass multiple configuration options to NewProducer.

func NewProducerConfig() *ProducerConfig

func (config *ProducerConfig) Validate() error

type RandomPartitioner ¶

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

RandomPartitioner implements the Partitioner interface by choosing a random partition each time.

func (p *RandomPartitioner) Partition(key Encoder, numPartitions int32) int32

type RequiredAcks ¶

type RequiredAcks int16

RequiredAcks is used in Produce Requests to tell the broker how many replica acknowledgements it must see before responding. Any positive int16 value is valid, or the constants defined here.

const (
	// NoResponse doesn't send any response, the TCP ACK is all you get.
	NoResponse RequiredAcks = 0
	// WaitForLocal waits for only the local commit to succeed before responding.
	WaitForLocal RequiredAcks = 1
	// WaitForAll waits for all replicas to commit before responding.
	WaitForAll RequiredAcks = -1
)

type RoundRobinPartitioner ¶

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

RoundRobinPartitioner implements the Partitioner interface by walking through the available partitions one at a time.

func (p *RoundRobinPartitioner) Partition(key Encoder, numPartitions int32) int32

type SimpleProducer ¶

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

SimpleProducer publishes Kafka messages. It routes messages to the correct broker, refreshing metadata as appropriate, and parses responses for errors. You must call Close() on a producer to avoid leaks, it may not be garbage-collected automatically when it passes out of scope (this is in addition to calling Close on the underlying client, which is still necessary).

client, err := NewClient("client_id", []string{"localhost:9092"}, NewClientConfig())
if err != nil {
	panic(err)
} else {
	fmt.Println("> connected")
}
defer client.Close()

producer, err := NewSimpleProducer(client, "my_topic", nil)
if err != nil {
	panic(err)
}
defer producer.Close()

for {
	err = producer.SendMessage(nil, StringEncoder("testing 123"))
	if err != nil {
		panic(err)
	} else {
		fmt.Println("> message sent")
	}
}

Output:

func NewSimpleProducer(client *Client, topic string, partitioner PartitionerConstructor) (*SimpleProducer, error)

func (sp *SimpleProducer) Close() error

func (sp *SimpleProducer) SendMessage(key, value Encoder) error

type StdLogger ¶

type StdLogger interface {
	Print(v ...interface{})
	Printf(format string, v ...interface{})
	Println(v ...interface{})
}

StdLogger is used to log error messages.

var Logger StdLogger = log.New(ioutil.Discard, "[Sarama] ", log.LstdFlags)

type StringEncoder ¶

type StringEncoder string

StringEncoder implements the Encoder interface for Go strings so that you can do things like

producer.SendMessage(nil, sarama.StringEncoder("hello world"))

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

func (s StringEncoder) Length() int

type TestState ¶

type TestState interface {
	Error(args ...interface{})
	Fatal(args ...interface{})
	Fatalf(format string, args ...interface{})
}

TestState is a generic interface for a test state, implemented e.g. by testing.T

type TopicMetadata ¶

type TopicMetadata struct {
	Err        KError
	Name       string
	Partitions []*PartitionMetadata
}