sarama

package module

Go to main page

Versions in this module

v1

v1.20.1

Jan 10, 2019

Changes in this version

type ConfigEntry

+ Source ConfigSource

+ Synonyms []*ConfigSynonym

+ type ConfigSource int8

+ const SourceDefault

+ const SourceDynamicBroker

+ const SourceDynamicDefaultBroker

+ const SourceStaticBroker

+ const SourceTopic

+ const SourceUnknown

+ func (s ConfigSource) String() string

+ type ConfigSynonym struct

+ ConfigName string

+ ConfigValue string

+ Source ConfigSource

type DescribeConfigsRequest

+ IncludeSynonyms bool

+ Version int16

type DescribeConfigsResponse

+ Version int16

type OffsetRequest

+ func (r *OffsetRequest) ReplicaID() int32

+ func (r *OffsetRequest) SetReplicaID(id int32)

v1.20.0

Dec 10, 2018

Changes in this version

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

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

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

type Broker

+ func (b *Broker) Rack() string

type Client

+ InitProducerID func() (*InitProducerIDResponse, error)

type CompressionCodec

+ const CompressionZSTD

type KError

+ const ErrDelegationTokenAuthDisabled

+ const ErrDelegationTokenAuthorizationFailed

+ const ErrDelegationTokenExpired

+ const ErrDelegationTokenNotFound

+ const ErrDelegationTokenOwnerMismatch

+ const ErrDelegationTokenRequestNotAllowed

+ const ErrFetchSessionIDNotFound

+ const ErrGroupIDNotFound

+ const ErrInvalidFetchSessionEpoch

+ const ErrInvalidPrincipalType

+ const ErrListenerNotFound

+ const ErrNonEmptyGroup

type OffsetCommitResponse

+ ThrottleTimeMs int32

+ Version int16

type OffsetFetchRequest

+ func (r *OffsetFetchRequest) ZeroPartitions()

type OffsetFetchResponse

+ Err KError

+ ThrottleTimeMs int32

+ Version int16

type OffsetFetchResponseBlock

+ LeaderEpoch int32

v1.19.0

Sep 27, 2018

Changes in this version

+ var BalanceStrategyRange = &balanceStrategy

+ var BalanceStrategyRoundRobin = &balanceStrategy

+ var ErrClosedConsumerGroup = errors.New("kafka: tried to use a consumer group that was closed")

+ type BalanceStrategy interface

+ Name func() string

+ Plan func(members map[string]ConsumerGroupMemberMetadata, topics map[string][]int32) (BalanceStrategyPlan, error)

+ type BalanceStrategyPlan map[string]map[string][]int32

+ func (p BalanceStrategyPlan) Add(memberID, topic string, partitions ...int32)

+ type ConsumerGroup interface

+ Close func() error

+ Consume func(ctx context.Context, topics []string, handler ConsumerGroupHandler) error

+ Errors func() <-chan error

+ func NewConsumerGroup(addrs []string, groupID string, config *Config) (ConsumerGroup, error)

+ func NewConsumerGroupFromClient(groupID string, client Client) (ConsumerGroup, error)

+ type ConsumerGroupClaim interface

+ HighWaterMarkOffset func() int64

+ InitialOffset func() int64

+ Messages func() <-chan *ConsumerMessage

+ Partition func() int32

+ Topic func() string

+ type ConsumerGroupHandler interface

+ Cleanup func(ConsumerGroupSession) error

+ ConsumeClaim func(ConsumerGroupSession, ConsumerGroupClaim) error

+ Setup func(ConsumerGroupSession) error

+ type ConsumerGroupSession interface

+ Claims func() map[string][]int32

+ Context func() context.Context

+ GenerationID func() int32

+ MarkMessage func(msg *ConsumerMessage, metadata string)

+ MarkOffset func(topic string, partition int32, offset int64, metadata string)

+ MemberID func() string

+ ResetOffset func(topic string, partition int32, offset int64, metadata string)

v1.18.0

Sep 7, 2018

Changes in this version

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

+ type ClusterAdmin interface

+ AlterConfig func(resourceType ConfigResourceType, name string, entries map[string]*string, ...) error

+ Close func() error

+ CreateACL func(resource Resource, acl Acl) error

+ CreatePartitions func(topic string, count int32, assignment [][]int32, validateOnly bool) error

+ CreateTopic func(topic string, detail *TopicDetail, validateOnly bool) error

+ DeleteACL func(filter AclFilter, validateOnly bool) ([]MatchingAcl, error)

+ DeleteRecords func(topic string, partitionOffsets map[int32]int64) error

+ DeleteTopic func(topic string) error

+ DescribeConfig func(resource ConfigResource) ([]ConfigEntry, error)

+ ListAcls func(filter AclFilter) ([]ResourceAcls, error)

+ func NewClusterAdmin(addrs []string, conf *Config) (ClusterAdmin, error)

type Config

+ Admin struct{ ... }

+ type DynamicConsistencyPartitioner interface

+ MessageRequiresConsistency func(message *ProducerMessage) bool

+ type HashPartitionerOption func(*hashPartitioner)

+ func WithAbsFirst() HashPartitionerOption

+ func WithCustomFallbackPartitioner(randomHP *hashPartitioner) HashPartitionerOption

+ func WithCustomHashFunction(hasher func() hash.Hash32) HashPartitionerOption

type MessageSet

+ OverflowMessage bool

+ type MockAlterConfigsResponse struct

+ func NewMockAlterConfigsResponse(t TestReporter) *MockAlterConfigsResponse

+ func (mr *MockAlterConfigsResponse) For(reqBody versionedDecoder) encoder

+ type MockCreateAclsResponse struct

+ func NewMockCreateAclsResponse(t TestReporter) *MockCreateAclsResponse

+ func (mr *MockCreateAclsResponse) For(reqBody versionedDecoder) encoder

+ type MockCreatePartitionsResponse struct

+ func NewMockCreatePartitionsResponse(t TestReporter) *MockCreatePartitionsResponse

+ func (mr *MockCreatePartitionsResponse) For(reqBody versionedDecoder) encoder

+ type MockCreateTopicsResponse struct

+ func NewMockCreateTopicsResponse(t TestReporter) *MockCreateTopicsResponse

+ func (mr *MockCreateTopicsResponse) For(reqBody versionedDecoder) encoder

+ type MockDeleteAclsResponse struct

+ func NewMockDeleteAclsResponse(t TestReporter) *MockDeleteAclsResponse

+ func (mr *MockDeleteAclsResponse) For(reqBody versionedDecoder) encoder

+ type MockDeleteRecordsResponse struct

+ func NewMockDeleteRecordsResponse(t TestReporter) *MockDeleteRecordsResponse

+ func (mr *MockDeleteRecordsResponse) For(reqBody versionedDecoder) encoder

+ type MockDeleteTopicsResponse struct

+ func NewMockDeleteTopicsResponse(t TestReporter) *MockDeleteTopicsResponse

+ func (mr *MockDeleteTopicsResponse) For(reqBody versionedDecoder) encoder

+ type MockDescribeConfigsResponse struct

+ func NewMockDescribeConfigsResponse(t TestReporter) *MockDescribeConfigsResponse

+ func (mr *MockDescribeConfigsResponse) For(reqBody versionedDecoder) encoder

+ type MockListAclsResponse struct

+ func NewMockListAclsResponse(t TestReporter) *MockListAclsResponse

+ func (mr *MockListAclsResponse) For(reqBody versionedDecoder) encoder

type Partitioner

+ func NewReferenceHashPartitioner(topic string) Partitioner

type PartitionerConstructor

+ func NewCustomPartitioner(options ...HashPartitionerOption) PartitionerConstructor

v1.17.0

May 30, 2018

Changes in this version

+ const CompressionLevelDefault

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

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

+ var MaxVersion = V1_1_0_0

+ var MinVersion = V0_8_2_0

+ var NoNode = &Broker

+ var SupportedVersions = []KafkaVersion

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

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

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

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

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

type Broker

+ func (b *Broker) CreatePartitions(request *CreatePartitionsRequest) (*CreatePartitionsResponse, error)

+ func (b *Broker) DeleteGroups(request *DeleteGroupsRequest) (*DeleteGroupsResponse, error)

+ func (b *Broker) DeleteRecords(request *DeleteRecordsRequest) (*DeleteRecordsResponse, error)

+ func (b *Broker) FindCoordinator(request *FindCoordinatorRequest) (*FindCoordinatorResponse, error)

type Client

+ Controller func() (*Broker, error)

type CompressionCodec

+ func (cc CompressionCodec) String() string

+ type CoordinatorType int8

+ const CoordinatorGroup

+ const CoordinatorTransaction

+ type DeleteGroupsRequest struct

+ Groups []string

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

+ type DeleteGroupsResponse struct

+ GroupErrorCodes map[string]KError

+ ThrottleTime time.Duration

+ type DeleteRecordsRequest struct

+ Timeout time.Duration

+ Topics map[string]*DeleteRecordsRequestTopic

+ type DeleteRecordsRequestTopic struct

+ PartitionOffsets map[int32]int64

+ type DeleteRecordsResponse struct

+ ThrottleTime time.Duration

+ Topics map[string]*DeleteRecordsResponseTopic

+ Version int16

+ type DeleteRecordsResponsePartition struct

+ Err KError

+ LowWatermark int64

+ type DeleteRecordsResponseTopic struct

+ Partitions map[int32]*DeleteRecordsResponsePartition

type DeleteTopicsRequest

+ Version int16

+ type FindCoordinatorRequest struct

+ CoordinatorKey string

+ CoordinatorType CoordinatorType

+ Version int16

+ type FindCoordinatorResponse struct

+ Coordinator *Broker

+ Err KError

+ ErrMsg *string

+ ThrottleTime time.Duration

+ Version int16

type JoinGroupRequest

+ RebalanceTimeout int32

+ Version int16

type JoinGroupResponse

+ ThrottleTime int32

+ Version int16

type Message

+ CompressionLevel int

type MetadataRequest

+ AllowAutoTopicCreation bool

+ Version int16

type MetadataResponse

+ ClusterID *string

+ ControllerID int32

+ ThrottleTimeMs int32

+ Version int16

+ type MockFindCoordinatorResponse struct

+ func NewMockFindCoordinatorResponse(t TestReporter) *MockFindCoordinatorResponse

+ func (mr *MockFindCoordinatorResponse) For(reqBody versionedDecoder) encoder

+ func (mr *MockFindCoordinatorResponse) SetCoordinator(coordinatorType CoordinatorType, group string, broker *MockBroker) *MockFindCoordinatorResponse

+ func (mr *MockFindCoordinatorResponse) SetError(coordinatorType CoordinatorType, group string, kerror KError) *MockFindCoordinatorResponse

type MockMetadataResponse

+ func (mmr *MockMetadataResponse) SetController(brokerID int32) *MockMetadataResponse

type OffsetCommitRequest

+ func (r *OffsetCommitRequest) Offset(topic string, partitionID int32) (int64, string, error)

type PartitionMetadata

+ OfflineReplicas []int32

type RecordBatch

+ CompressionLevel int

type Records

+ MsgSet *MessageSet

+ RecordBatch *RecordBatch

type TopicMetadata

+ IsInternal bool

v1.16.0

Feb 12, 2018

Changes in this version

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

+ type Acl struct

+ Host string

+ Operation AclOperation

+ PermissionType AclPermissionType

+ Principal string

+ type AclCreation struct

+ type AclCreationResponse struct

+ Err KError

+ ErrMsg *string

+ type AclFilter struct

+ Host *string

+ Operation AclOperation

+ PermissionType AclPermissionType

+ Principal *string

+ ResourceName *string

+ ResourceType AclResourceType

+ type AclOperation int

+ const AclOperationAll

+ const AclOperationAlter

+ const AclOperationAlterConfigs

+ const AclOperationAny

+ const AclOperationClusterAction

+ const AclOperationCreate

+ const AclOperationDelete

+ const AclOperationDescribe

+ const AclOperationDescribeConfigs

+ const AclOperationIdempotentWrite

+ const AclOperationRead

+ const AclOperationUnknown

+ const AclOperationWrite

+ type AclPermissionType int

+ const AclPermissionAllow

+ const AclPermissionAny

+ const AclPermissionDeny

+ const AclPermissionUnknown

+ type AclResourceType int

+ const AclResourceAny

+ const AclResourceCluster

+ const AclResourceGroup

+ const AclResourceTopic

+ const AclResourceTransactionalID

+ const AclResourceUnknown

+ type AddOffsetsToTxnRequest struct

+ GroupID string

+ ProducerEpoch int16

+ ProducerID int64

+ TransactionalID string

+ type AddOffsetsToTxnResponse struct

+ Err KError

+ ThrottleTime time.Duration

+ type AddPartitionsToTxnRequest struct

+ ProducerEpoch int16

+ ProducerID int64

+ TopicPartitions map[string][]int32

+ TransactionalID string

+ type AddPartitionsToTxnResponse struct

+ Errors map[string][]*PartitionError

+ ThrottleTime time.Duration

+ type AlterConfigsRequest struct

+ Resources []*AlterConfigsResource

+ ValidateOnly bool

+ type AlterConfigsResource struct

+ ConfigEntries map[string]*string

+ Name string

+ Type ConfigResourceType

+ type AlterConfigsResourceResponse struct

+ ErrorCode int16

+ ErrorMsg string

+ Name string

+ Type ConfigResourceType

+ type AlterConfigsResponse struct

+ Resources []*AlterConfigsResourceResponse

+ ThrottleTime time.Duration

type Broker

+ func (b *Broker) AddOffsetsToTxn(request *AddOffsetsToTxnRequest) (*AddOffsetsToTxnResponse, error)

+ func (b *Broker) AddPartitionsToTxn(request *AddPartitionsToTxnRequest) (*AddPartitionsToTxnResponse, error)

+ func (b *Broker) AlterConfigs(request *AlterConfigsRequest) (*AlterConfigsResponse, error)

+ func (b *Broker) CreateAcls(request *CreateAclsRequest) (*CreateAclsResponse, error)

+ func (b *Broker) CreateTopics(request *CreateTopicsRequest) (*CreateTopicsResponse, error)

+ func (b *Broker) DeleteAcls(request *DeleteAclsRequest) (*DeleteAclsResponse, error)

+ func (b *Broker) DeleteTopics(request *DeleteTopicsRequest) (*DeleteTopicsResponse, error)

+ func (b *Broker) DescribeAcls(request *DescribeAclsRequest) (*DescribeAclsResponse, error)

+ func (b *Broker) DescribeConfigs(request *DescribeConfigsRequest) (*DescribeConfigsResponse, error)

+ func (b *Broker) EndTxn(request *EndTxnRequest) (*EndTxnResponse, error)

+ func (b *Broker) InitProducerID(request *InitProducerIDRequest) (*InitProducerIDResponse, error)

+ func (b *Broker) TxnOffsetCommit(request *TxnOffsetCommitRequest) (*TxnOffsetCommitResponse, error)

+ type ConfigEntry struct

+ Default bool

+ Name string

+ ReadOnly bool

+ Sensitive bool

+ Value string

+ type ConfigResource struct

+ ConfigNames []string

+ Name string

+ Type ConfigResourceType

+ type ConfigResourceType int8

+ const AnyResource

+ const BrokerResource

+ const ClusterResource

+ const GroupResource

+ const TopicResource

+ const UnknownResource

+ type CreateAclsRequest struct

+ AclCreations []*AclCreation

+ type CreateAclsResponse struct

+ AclCreationResponses []*AclCreationResponse

+ ThrottleTime time.Duration

+ type CreateTopicsRequest struct

+ Timeout time.Duration

+ TopicDetails map[string]*TopicDetail

+ ValidateOnly bool

+ Version int16

+ type CreateTopicsResponse struct

+ ThrottleTime time.Duration

+ TopicErrors map[string]*TopicError

+ Version int16

+ type DeleteAclsRequest struct

+ Filters []*AclFilter

+ type DeleteAclsResponse struct

+ FilterResponses []*FilterResponse

+ ThrottleTime time.Duration

+ type DeleteTopicsRequest struct

+ Timeout time.Duration

+ Topics []string

+ type DeleteTopicsResponse struct

+ ThrottleTime time.Duration

+ TopicErrorCodes map[string]KError

+ Version int16

+ type DescribeAclsRequest struct

+ type DescribeAclsResponse struct

+ Err KError

+ ErrMsg *string

+ ResourceAcls []*ResourceAcls

+ ThrottleTime time.Duration

+ type DescribeConfigsRequest struct

+ Resources []*ConfigResource

+ type DescribeConfigsResponse struct

+ Resources []*ResourceResponse

+ ThrottleTime time.Duration

+ type EndTxnRequest struct

+ ProducerEpoch int16

+ ProducerID int64

+ TransactionResult bool

+ TransactionalID string

+ type EndTxnResponse struct

+ Err KError

+ ThrottleTime time.Duration

type FetchResponse

+ func (r *FetchResponse) SetLastOffsetDelta(topic string, partition int32, offset int32)

type FetchResponseBlock

+ Partial bool

+ RecordsSet []*Records

+ type FilterResponse struct

+ Err KError

+ ErrMsg *string

+ MatchingAcls []*MatchingAcl

+ type InitProducerIDRequest struct

+ TransactionTimeout time.Duration

+ TransactionalID *string

+ type InitProducerIDResponse struct

+ Err KError

+ ProducerEpoch int16

+ ProducerID int64

+ ThrottleTime time.Duration

+ type MatchingAcl struct

+ Err KError

+ ErrMsg *string

type MockBroker

+ func NewMockBrokerListener(t TestReporter, brokerID int32, listener net.Listener) *MockBroker

type MockProduceResponse

+ func (mr *MockProduceResponse) SetVersion(version int16) *MockProduceResponse

+ type PartitionError struct

+ Err KError

+ Partition int32

+ type PartitionOffsetMetadata struct

+ Metadata *string

+ Offset int64

+ Partition int32

+ type Resource struct

+ ResourceName string

+ ResourceType AclResourceType

+ type ResourceAcls struct

+ Acls []*Acl

+ type ResourceResponse struct

+ Configs []*ConfigEntry

+ ErrorCode int16

+ ErrorMsg string

+ Name string

+ Type ConfigResourceType

+ type TopicDetail struct

+ ConfigEntries map[string]*string

+ NumPartitions int32

+ ReplicaAssignment map[int32][]int32

+ ReplicationFactor int16

+ type TopicError struct

+ Err KError

+ ErrMsg *string

+ type TxnOffsetCommitRequest struct

+ GroupID string

+ ProducerEpoch int16

+ ProducerID int64

+ Topics map[string][]*PartitionOffsetMetadata

+ TransactionalID string

+ type TxnOffsetCommitResponse struct

+ ThrottleTime time.Duration

+ Topics map[string][]*PartitionError

v1.15.0

Dec 8, 2017

Changes in this version

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

+ type CreatePartitionsRequest struct

+ Timeout time.Duration

+ TopicPartitions map[string]*TopicPartition

+ ValidateOnly bool

+ type CreatePartitionsResponse struct

+ ThrottleTime time.Duration

+ TopicPartitionErrors map[string]*TopicPartitionError

type KError

+ const ErrConcurrentTransactions

+ const ErrDuplicateSequenceNumber

+ const ErrInvalidProducerEpoch

+ const ErrInvalidProducerIDMapping

+ const ErrInvalidTransactionTimeout

+ const ErrInvalidTxnState

+ const ErrKafkaStorageError

+ const ErrLogDirNotFound

+ const ErrOperationNotAttempted

+ const ErrOutOfOrderSequenceNumber

+ const ErrReassignmentInProgress

+ const ErrSASLAuthenticationFailed

+ const ErrSecurityDisabled

+ const ErrTransactionCoordinatorFenced

+ const ErrTransactionalIDAuthorizationFailed

+ const ErrUnknownProducerID

type KafkaVersion

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

+ func (v KafkaVersion) String() string

+ type TopicPartition struct

+ Assignment [][]int32

+ Count int32

+ type TopicPartitionError struct

+ Err KError

+ ErrMsg *string

v1.14.0

Nov 13, 2017

Changes in this version

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

+ type AbortedTransaction struct

+ FirstOffset int64

+ ProducerID int64

type ConsumerMessage

+ Headers []*RecordHeader

type FetchRequest

+ Isolation IsolationLevel

type FetchResponse

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

+ func (r *FetchResponse) SetLastStableOffset(topic string, partition int32, offset int64)

type FetchResponseBlock

+ AbortedTransactions []*AbortedTransaction

+ LastStableOffset int64

+ Records Records

+ type IsolationLevel int8

+ const ReadCommitted

+ const ReadUncommitted

type MockOffsetResponse

+ func (mor *MockOffsetResponse) SetVersion(version int16) *MockOffsetResponse

type ProduceRequest

+ TransactionalID *string

+ func (r *ProduceRequest) AddBatch(topic string, partition int32, batch *RecordBatch)

type ProducerMessage

+ Headers []RecordHeader

+ type Record struct

+ Attributes int8

+ Headers []*RecordHeader

+ Key []byte

+ OffsetDelta int64

+ TimestampDelta time.Duration

+ Value []byte

+ type RecordBatch struct

+ Codec CompressionCodec

+ Control bool

+ FirstOffset int64

+ FirstSequence int32

+ FirstTimestamp time.Time

+ LastOffsetDelta int32

+ MaxTimestamp time.Time

+ PartialTrailingRecord bool

+ PartitionLeaderEpoch int32

+ ProducerEpoch int16

+ ProducerID int64

+ Records []*Record

+ Version int8

+ type RecordHeader struct

+ Key []byte

+ Value []byte

+ type Records struct

+ type Timestamp struct

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_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 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

type PartitionerConstructor

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

v1.11.0

Dec 20, 2016

Changes in this version

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

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

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 KError

+ 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 RequestNotifierFunc func(bytesRead, bytesWritten int)

v1.10.1

Aug 30, 2016

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

+ 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 TestReporter interface

+ Error func(...interface{})

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

+ Fatal func(...interface{})

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

v1.8.0

Feb 1, 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 JoinGroupRequest

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

type JoinGroupResponse

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

type SyncGroupRequest

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

type SyncGroupResponse

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

v1.7.0

Dec 11, 2015

Changes in this version

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

+ type JoinGroupResponse struct

+ Err KError

+ GenerationId int32

+ GroupProtocol string

+ LeaderId string

+ MemberId string

+ Members map[string][]byte

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

+ GenerationId int32

+ GroupAssignments map[string][]byte

+ GroupId string

+ MemberId string

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

+ type SyncGroupResponse struct

+ Err KError

+ MemberAssignment []byte

v1.6.1

Sep 25, 2015

v1.6.0

Sep 4, 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 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

v1.4.3

Jul 21, 2015

v1.4.2

May 27, 2015

v1.4.1

May 25, 2015

v1.4.0

May 1, 2015

Changes in this version

type Consumer

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

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

type PartitionConsumer

+ HighWaterMarkOffset func() int64

v1.3.0

Apr 16, 2015

Changes in this version

type Client

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

+ RefreshCoordinator func(consumerGroup string) error

type ConsumerMetadataResponse

+ Coordinator *Broker

v1.2.0

Apr 7, 2015

Changes in this version

type OffsetCommitRequest

+ ConsumerGroupGeneration int32

+ ConsumerID string

+ RetentionTime int64

+ Version int16

type OffsetFetchRequest

+ Version int16

v1.1.0

Mar 20, 2015

Changes in this version

type MetadataResponse

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

v1.0.0

Mar 17, 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) AddTopicPartition(topic string, partition, brokerID int32, replicas, isr []int32, err KError)

+ type OffsetCommitRequest struct

+ ConsumerGroup string

+ 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

+ 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