sarama

package module

Go to main page

Versions in this module

v1

v1.13.0

Oct 4, 2017

Changes in this version

type ConsumerMessage

+ BlockTimestamp time.Time

type FetchRequest

+ MaxBytes int32

type MockFetchResponse

+ func (mfr *MockFetchResponse) SetVersion(version int16) *MockFetchResponse

type PartitionOffsetManager

+ ResetOffset func(offset int64, metadata string)

v1.12.0

May 8, 2017

Changes in this version

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

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

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

type Broker

+ func (b *Broker) ApiVersions(request *ApiVersionsRequest) (*ApiVersionsResponse, error)

type Client

+ Brokers func() []*Broker

+ InSyncReplicas func(topic string, partitionID int32) ([]int32, error)

type CompressionCodec

+ const CompressionLZ4

type Config

+ MetricRegistry metrics.Registry

type Consumer

+ HighWaterMarks func() map[string]map[int32]int64

type GroupMemberDescription

+ func (gmd *GroupMemberDescription) GetMemberAssignment() (*ConsumerGroupMemberAssignment, error)

+ func (gmd *GroupMemberDescription) GetMemberMetadata() (*ConsumerGroupMemberMetadata, error)

+ type GroupProtocol struct

+ Metadata []byte

+ Name string

type JoinGroupRequest

+ OrderedGroupProtocols []*GroupProtocol

type KError

+ const ErrInvalidConfig

+ const ErrInvalidPartitions

+ const ErrInvalidReplicaAssignment

+ const ErrInvalidReplicationFactor

+ const ErrInvalidRequest

+ const ErrNotController

+ const ErrPolicyViolation

+ const ErrTopicAlreadyExists

+ const ErrUnsupportedForMessageFormat

type MockBroker

+ func (b *MockBroker) SetNotifier(notifier RequestNotifierFunc)

type OffsetRequest

+ Version int16

type OffsetResponse

+ Version int16

type OffsetResponseBlock

+ Offset int64

+ Timestamp int64

type PartitionerConstructor

+ func NewCustomHashPartitioner(hasher func() hash.Hash32) PartitionerConstructor

+ type RequestNotifierFunc func(bytesRead, bytesWritten int)

v1.10.0

Aug 2, 2016

Changes in this version

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

+ 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)

+ type ApiVersionsRequest struct

+ type ApiVersionsResponse struct

+ ApiVersions []*ApiVersionsResponseBlock

+ Err KError

+ type ApiVersionsResponseBlock struct

+ ApiKey int16

+ MaxVersion int16

+ MinVersion int16

type Config

+ Version KafkaVersion

type ConsumerMessage

+ Timestamp time.Time

type FetchRequest

+ Version int16

type FetchResponse

+ ThrottleTime time.Duration

+ Version int16

type KError

+ const ErrIllegalSASLState

+ const ErrInvalidTimestamp

+ const ErrNetworkException

+ const ErrUnsupportedSASLMechanism

+ const ErrUnsupportedVersion

+ type KafkaVersion struct

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

type Message

+ Timestamp time.Time

+ Version int8

type ProduceRequest

+ Version int16

type ProduceResponse

+ ThrottleTime time.Duration

+ Version int16

type ProduceResponseBlock

+ Timestamp time.Time

type ProducerMessage

+ Timestamp time.Time

+ type SaslHandshakeRequest struct

+ Mechanism string

+ type SaslHandshakeResponse struct

+ EnabledMechanisms []string

+ Err KError

type SyncProducer

+ SendMessages func(msgs []*ProducerMessage) error

v1.9.0

May 16, 2016

Changes in this version

+ const GroupGenerationUndefined

type Broker

+ func (b *Broker) DescribeGroups(request *DescribeGroupsRequest) (*DescribeGroupsResponse, error)

+ func (b *Broker) Heartbeat(request *HeartbeatRequest) (*HeartbeatResponse, error)

+ func (b *Broker) JoinGroup(request *JoinGroupRequest) (*JoinGroupResponse, error)

+ func (b *Broker) LeaveGroup(request *LeaveGroupRequest) (*LeaveGroupResponse, error)

+ func (b *Broker) ListGroups(request *ListGroupsRequest) (*ListGroupsResponse, error)

+ func (b *Broker) SyncGroup(request *SyncGroupRequest) (*SyncGroupResponse, error)

+ type ConsumerGroupMemberAssignment struct

+ Topics map[string][]int32

+ UserData []byte

+ Version int16

+ type ConsumerGroupMemberMetadata struct

+ Topics []string

+ UserData []byte

+ Version int16

+ type DescribeGroupsRequest struct

+ Groups []string

+ func (r *DescribeGroupsRequest) AddGroup(group string)

+ type DescribeGroupsResponse struct

+ Groups []*GroupDescription

+ type GroupDescription struct

+ Err KError

+ GroupId string

+ Members map[string]*GroupMemberDescription

+ Protocol string

+ ProtocolType string

+ State string

+ type GroupMemberDescription struct

+ ClientHost string

+ ClientId string

+ MemberAssignment []byte

+ MemberMetadata []byte

+ type HeartbeatRequest struct

+ GenerationId int32

+ GroupId string

+ MemberId string

+ type HeartbeatResponse struct

+ Err KError

+ type JoinGroupRequest struct

+ GroupId string

+ GroupProtocols map[string][]byte

+ MemberId string

+ ProtocolType string

+ SessionTimeout int32

+ func (r *JoinGroupRequest) AddGroupProtocol(name string, metadata []byte)

+ func (r *JoinGroupRequest) AddGroupProtocolMetadata(name string, metadata *ConsumerGroupMemberMetadata) error

+ type JoinGroupResponse struct

+ Err KError

+ GenerationId int32

+ GroupProtocol string

+ LeaderId string

+ MemberId string

+ Members map[string][]byte

+ func (r *JoinGroupResponse) GetMembers() (map[string]ConsumerGroupMemberMetadata, error)

type KError

+ const ErrClusterAuthorizationFailed

+ const ErrGroupAuthorizationFailed

+ const ErrIllegalGeneration

+ const ErrInconsistentGroupProtocol

+ const ErrInvalidCommitOffsetSize

+ const ErrInvalidGroupId

+ const ErrInvalidRequiredAcks

+ const ErrInvalidSessionTimeout

+ const ErrRebalanceInProgress

+ const ErrTopicAuthorizationFailed

+ const ErrUnknownMemberId

+ type LeaveGroupRequest struct

+ GroupId string

+ MemberId string

+ type LeaveGroupResponse struct

+ Err KError

+ type ListGroupsRequest struct

+ type ListGroupsResponse struct

+ Err KError

+ Groups map[string]string

+ type MockBroker struct

+ func NewMockBroker(t TestReporter, brokerID int32) *MockBroker

+ func NewMockBrokerAddr(t TestReporter, brokerID int32, addr string) *MockBroker

+ func (b *MockBroker) Addr() string

+ func (b *MockBroker) BrokerID() int32

+ func (b *MockBroker) Close()

+ func (b *MockBroker) History() []RequestResponse

+ func (b *MockBroker) Port() int32

+ func (b *MockBroker) Returns(e encoder)

+ func (b *MockBroker) SetHandlerByMap(handlerMap map[string]MockResponse)

+ func (b *MockBroker) SetLatency(latency time.Duration)

+ type MockConsumerMetadataResponse struct

+ func NewMockConsumerMetadataResponse(t TestReporter) *MockConsumerMetadataResponse

+ func (mr *MockConsumerMetadataResponse) For(reqBody decoder) encoder

+ func (mr *MockConsumerMetadataResponse) SetCoordinator(group string, broker *MockBroker) *MockConsumerMetadataResponse

+ func (mr *MockConsumerMetadataResponse) SetError(group string, kerror KError) *MockConsumerMetadataResponse

+ type MockFetchResponse struct

+ func NewMockFetchResponse(t TestReporter, batchSize int) *MockFetchResponse

+ func (mfr *MockFetchResponse) For(reqBody decoder) encoder

+ func (mfr *MockFetchResponse) SetHighWaterMark(topic string, partition int32, offset int64) *MockFetchResponse

+ func (mfr *MockFetchResponse) SetMessage(topic string, partition int32, offset int64, msg Encoder) *MockFetchResponse

+ type MockMetadataResponse struct

+ func NewMockMetadataResponse(t TestReporter) *MockMetadataResponse

+ func (mmr *MockMetadataResponse) For(reqBody decoder) encoder

+ func (mmr *MockMetadataResponse) SetBroker(addr string, brokerID int32) *MockMetadataResponse

+ func (mmr *MockMetadataResponse) SetLeader(topic string, partition, brokerID int32) *MockMetadataResponse

+ type MockOffsetCommitResponse struct

+ func NewMockOffsetCommitResponse(t TestReporter) *MockOffsetCommitResponse

+ func (mr *MockOffsetCommitResponse) For(reqBody decoder) encoder

+ func (mr *MockOffsetCommitResponse) SetError(group, topic string, partition int32, kerror KError) *MockOffsetCommitResponse

+ type MockOffsetFetchResponse struct

+ func NewMockOffsetFetchResponse(t TestReporter) *MockOffsetFetchResponse

+ func (mr *MockOffsetFetchResponse) For(reqBody decoder) encoder

+ func (mr *MockOffsetFetchResponse) SetOffset(group, topic string, partition int32, offset int64, metadata string, ...) *MockOffsetFetchResponse

+ type MockOffsetResponse struct

+ func NewMockOffsetResponse(t TestReporter) *MockOffsetResponse

+ func (mor *MockOffsetResponse) For(reqBody decoder) encoder

+ func (mor *MockOffsetResponse) SetOffset(topic string, partition int32, time, offset int64) *MockOffsetResponse

+ type MockProduceResponse struct

+ func NewMockProduceResponse(t TestReporter) *MockProduceResponse

+ func (mr *MockProduceResponse) For(reqBody decoder) encoder

+ func (mr *MockProduceResponse) SetError(topic string, partition int32, kerror KError) *MockProduceResponse

+ type MockResponse interface

+ For func(reqBody decoder) (res encoder)

+ type MockSequence struct

+ func NewMockSequence(responses ...interface{}) *MockSequence

+ func (mc *MockSequence) For(reqBody decoder) (res encoder)

+ type MockWrapper struct

+ func NewMockWrapper(res encoder) *MockWrapper

+ func (mw *MockWrapper) For(reqBody decoder) (res encoder)

+ type RequestResponse struct

+ Request requestBody

+ Response encoder

+ type SyncGroupRequest struct

+ GenerationId int32

+ GroupAssignments map[string][]byte

+ GroupId string

+ MemberId string

+ func (r *SyncGroupRequest) AddGroupAssignment(memberId string, memberAssignment []byte)

+ func (r *SyncGroupRequest) AddGroupAssignmentMember(memberId string, memberAssignment *ConsumerGroupMemberAssignment) error

+ type SyncGroupResponse struct

+ Err KError

+ MemberAssignment []byte

+ func (r *SyncGroupResponse) GetMemberAssignment() (*ConsumerGroupMemberAssignment, error)

+ type TestReporter interface

+ Error func(...interface{})

+ Errorf func(string, ...interface{})

+ Fatal func(...interface{})

+ Fatalf func(string, ...interface{})

v1.6.1

Sep 25, 2015

Changes in this version

+ type OffsetManager interface

+ Close func() error

+ ManagePartition func(topic string, partition int32) (PartitionOffsetManager, error)

+ func NewOffsetManagerFromClient(group string, client Client) (OffsetManager, error)

+ type PartitionOffsetManager interface

+ AsyncClose func()

+ Close func() error

+ Errors func() <-chan *ConsumerError

+ MarkOffset func(offset int64, metadata string)

+ NextOffset func() (int64, string)

v1.5.0

Aug 17, 2015

Changes in this version

type Client

+ Coordinator func(consumerGroup string) (*Broker, error)

+ RefreshCoordinator func(consumerGroup string) error

type Consumer

+ Partitions func(topic string) ([]int32, error)

+ Topics func() ([]string, error)

type ConsumerMetadataResponse

+ Coordinator *Broker

type OffsetCommitResponse

+ func (r *OffsetCommitResponse) AddError(topic string, partition int32, kerror KError)

type OffsetFetchResponse

+ func (r *OffsetFetchResponse) AddBlock(topic string, partition int32, block *OffsetFetchResponseBlock)

+ func (r *OffsetFetchResponse) GetBlock(topic string, partition int32) *OffsetFetchResponseBlock

type PartitionConsumer

+ HighWaterMarkOffset func() int64

v1.2.0

Apr 7, 2015

Changes in this version

+ const OffsetNewest

+ const OffsetOldest

+ const ReceiveTime

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

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

+ 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 ErrNotConnected = errors.New("kafka: broker not connected")

+ var ErrOutOfBrokers = errors.New(...)

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

+ var MaxRequestSize int32 = 100 * 1024 * 1024

+ var MaxResponseSize int32 = 100 * 1024 * 1024

+ var PanicHandler func(interface{})

+ type AsyncProducer interface

+ AsyncClose func()

+ Close func() error

+ Errors func() <-chan *ProducerError

+ Input func() chan<- *ProducerMessage

+ Successes func() <-chan *ProducerMessage

+ func NewAsyncProducer(addrs []string, conf *Config) (AsyncProducer, error)

+ func NewAsyncProducerFromClient(client Client) (AsyncProducer, error)

+ type Broker struct

+ func NewBroker(addr string) *Broker

+ func (b *Broker) Addr() string

+ func (b *Broker) Close() error

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

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

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

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

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

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

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

+ func (b *Broker) ID() int32

+ func (b *Broker) Open(conf *Config) error

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

+ type ByteEncoder []byte

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

+ func (b ByteEncoder) Length() int

+ type Client interface

+ Close func() error

+ Closed func() bool

+ Config func() *Config

+ GetOffset func(topic string, partitionID int32, time int64) (int64, error)

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

+ Partitions func(topic string) ([]int32, error)

+ RefreshMetadata func(topics ...string) error

+ Replicas func(topic string, partitionID int32) ([]int32, error)

+ Topics func() ([]string, error)

+ WritablePartitions func(topic string) ([]int32, error)

+ func NewClient(addrs []string, conf *Config) (Client, error)

+ type CompressionCodec int8

+ const CompressionGZIP

+ const CompressionNone

+ const CompressionSnappy

+ type Config struct

+ ChannelBufferSize int

+ ClientID string

+ Consumer struct{ ... }

+ Metadata struct{ ... }

+ Net struct{ ... }

+ Producer struct{ ... }

+ func NewConfig() *Config

+ func (c *Config) Validate() error

+ type ConfigurationError string

+ func (err ConfigurationError) Error() string

+ type Consumer interface

+ Close func() error

+ ConsumePartition func(topic string, partition int32, offset int64) (PartitionConsumer, error)

+ func NewConsumer(addrs []string, config *Config) (Consumer, error)

+ func NewConsumerFromClient(client Client) (Consumer, error)

+ type ConsumerError struct

+ Err error

+ Partition int32

+ Topic string

+ func (ce ConsumerError) Error() string

+ type ConsumerErrors []*ConsumerError

+ func (ce ConsumerErrors) Error() string

+ type ConsumerMessage struct

+ Key []byte

+ Offset int64

+ Partition int32

+ Topic string

+ Value []byte

+ type ConsumerMetadataRequest struct

+ ConsumerGroup string

+ type ConsumerMetadataResponse struct

+ CoordinatorHost string

+ CoordinatorID int32

+ CoordinatorPort int32

+ Err KError

+ type Encoder interface

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

+ Length func() int

+ type FetchRequest struct

+ MaxWaitTime int32

+ MinBytes int32

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

+ type FetchResponse struct

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

+ func (fr *FetchResponse) AddError(topic string, partition int32, err KError)

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

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

+ type FetchResponseBlock struct

+ Err KError

+ HighWaterMarkOffset int64

+ MsgSet MessageSet

+ type KError int16

+ const ErrBrokerNotAvailable

+ const ErrConsumerCoordinatorNotAvailable

+ const ErrInvalidMessage

+ const ErrInvalidMessageSize

+ const ErrInvalidTopic

+ const ErrLeaderNotAvailable

+ const ErrMessageSetSizeTooLarge

+ const ErrMessageSizeTooLarge

+ const ErrNoError

+ const ErrNotCoordinatorForConsumer

+ const ErrNotEnoughReplicas

+ const ErrNotEnoughReplicasAfterAppend

+ const ErrNotLeaderForPartition

+ const ErrOffsetMetadataTooLarge

+ const ErrOffsetOutOfRange

+ const ErrOffsetsLoadInProgress

+ const ErrReplicaNotAvailable

+ const ErrRequestTimedOut

+ const ErrStaleControllerEpochCode

+ const ErrUnknown

+ const ErrUnknownTopicOrPartition

+ func (err KError) Error() string

+ type Message struct

+ Codec CompressionCodec

+ Key []byte

+ Set *MessageSet

+ Value []byte

+ type MessageBlock struct

+ Msg *Message

+ Offset int64

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

+ type MessageSet struct

+ Messages []*MessageBlock

+ PartialTrailingMessage bool

+ type MetadataRequest struct

+ Topics []string

+ type MetadataResponse struct

+ Brokers []*Broker

+ Topics []*TopicMetadata

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

+ func (m *MetadataResponse) AddTopic(topic string, err KError) *TopicMetadata

+ func (m *MetadataResponse) AddTopicPartition(topic string, partition, brokerID int32, replicas, isr []int32, err KError)

+ type OffsetCommitRequest struct

+ ConsumerGroup string

+ ConsumerGroupGeneration int32

+ ConsumerID string

+ RetentionTime int64

+ Version int16

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

+ type OffsetCommitResponse struct

+ Errors map[string]map[int32]KError

+ type OffsetFetchRequest struct

+ ConsumerGroup string

+ Version int16

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

+ type OffsetFetchResponse struct

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

+ type OffsetFetchResponseBlock struct

+ Err KError

+ Metadata string

+ Offset int64

+ type OffsetRequest struct

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

+ 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 struct

+ Err KError

+ Offsets []int64

+ type PacketDecodingError struct

+ Info string

+ func (err PacketDecodingError) Error() string

+ type PacketEncodingError struct

+ Info string

+ func (err PacketEncodingError) Error() string

+ type PartitionConsumer interface

+ AsyncClose func()

+ Close func() error

+ Errors func() <-chan *ConsumerError

+ Messages func() <-chan *ConsumerMessage

+ type PartitionMetadata struct

+ Err KError

+ ID int32

+ Isr []int32

+ Leader int32

+ Replicas []int32

+ type Partitioner interface

+ Partition func(message *ProducerMessage, numPartitions int32) (int32, error)

+ RequiresConsistency func() bool

+ func NewHashPartitioner(topic string) Partitioner

+ func NewManualPartitioner(topic string) Partitioner

+ func NewRandomPartitioner(topic string) Partitioner

+ func NewRoundRobinPartitioner(topic string) Partitioner

+ type PartitionerConstructor func(topic string) Partitioner

+ type ProduceRequest struct

+ RequiredAcks RequiredAcks

+ Timeout int32

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

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

+ 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 struct

+ Err KError

+ Offset int64

+ type ProducerError struct

+ Err error

+ Msg *ProducerMessage

+ func (pe ProducerError) Error() string

+ type ProducerErrors []*ProducerError

+ func (pe ProducerErrors) Error() string

+ type ProducerMessage struct

+ Key Encoder

+ Metadata interface{}

+ Offset int64

+ Partition int32

+ Topic string

+ Value Encoder

+ type RequiredAcks int16

+ const NoResponse

+ const WaitForAll

+ const WaitForLocal

+ type StdLogger interface

+ Print func(v ...interface{})

+ Printf func(format string, v ...interface{})

+ Println func(v ...interface{})

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

+ type StringEncoder string

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

+ func (s StringEncoder) Length() int

+ type SyncProducer interface

+ Close func() error

+ SendMessage func(msg *ProducerMessage) (partition int32, offset int64, err error)

+ func NewSyncProducer(addrs []string, config *Config) (SyncProducer, error)

+ func NewSyncProducerFromClient(client Client) (SyncProducer, error)

+ type TopicMetadata struct

+ Err KError

+ Name string

+ Partitions []*PartitionMetadata

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