kafka

package module

Go to main page

Versions in this module

v0

v0.4.32

Mar 9, 2022

Changes in this version

+ const SeekDontCheck

+ var ErrGenerationEnded = errors.New("consumer group generation has ended")

+ var ErrGroupClosed = errors.New("consumer group is closed")

+ func Marshal(v interface{}) ([]byte, error)

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

+ func TCP(address ...string) net.Addr

+ func Unmarshal(b []byte, v interface{}) error

+ type ACLEntry struct

+ Host string

+ Operation ACLOperationType

+ PermissionType ACLPermissionType

+ Principal string

+ ResourceName string

+ ResourcePatternType PatternType

+ ResourceType ResourceType

+ type ACLOperationType int8

+ const ACLOperationTypeAll

+ const ACLOperationTypeAlter

+ const ACLOperationTypeAlterConfigs

+ const ACLOperationTypeAny

+ const ACLOperationTypeClusterAction

+ const ACLOperationTypeCreate

+ const ACLOperationTypeDelete

+ const ACLOperationTypeDescribe

+ const ACLOperationTypeDescribeConfigs

+ const ACLOperationTypeIdempotentWrite

+ const ACLOperationTypeRead

+ const ACLOperationTypeUnknown

+ const ACLOperationTypeWrite

+ type ACLPermissionType int8

+ const ACLPermissionTypeAllow

+ const ACLPermissionTypeAny

+ const ACLPermissionTypeDeny

+ const ACLPermissionTypeUnknown

+ type AddOffsetsToTxnRequest struct

+ Addr net.Addr

+ GroupID string

+ ProducerEpoch int

+ ProducerID int

+ TransactionalID string

+ type AddOffsetsToTxnResponse struct

+ Error error

+ Throttle time.Duration

+ type AddPartitionToTxn struct

+ Partition int

+ type AddPartitionToTxnPartition struct

+ Error error

+ Partition int

+ type AddPartitionsToTxnRequest struct

+ Addr net.Addr

+ ProducerEpoch int

+ ProducerID int

+ Topics map[string][]AddPartitionToTxn

+ TransactionalID string

+ type AddPartitionsToTxnResponse struct

+ Throttle time.Duration

+ Topics map[string][]AddPartitionToTxnPartition

+ type AlterConfigRequestConfig struct

+ Name string

+ Value string

+ type AlterConfigRequestResource struct

+ Configs []AlterConfigRequestConfig

+ ResourceName string

+ ResourceType ResourceType

+ type AlterConfigsRequest struct

+ Addr net.Addr

+ Resources []AlterConfigRequestResource

+ ValidateOnly bool

+ type AlterConfigsResponse struct

+ Errors map[AlterConfigsResponseResource]error

+ Throttle time.Duration

+ type AlterConfigsResponseResource struct

+ Name string

+ Type int8

+ type AlterPartitionReassignmentsRequest struct

+ Addr net.Addr

+ Assignments []AlterPartitionReassignmentsRequestAssignment

+ Timeout time.Duration

+ Topic string

+ type AlterPartitionReassignmentsRequestAssignment struct

+ BrokerIDs []int

+ PartitionID int

+ type AlterPartitionReassignmentsResponse struct

+ Error error

+ PartitionResults []AlterPartitionReassignmentsResponsePartitionResult

+ type AlterPartitionReassignmentsResponsePartitionResult struct

+ Error error

+ PartitionID int

type ApiVersion

+ func (v ApiVersion) Format(w fmt.State, r rune)

+ type ApiVersionsRequest struct

+ Addr net.Addr

+ type ApiVersionsResponse struct

+ ApiKeys []ApiVersionsResponseApiKey

+ Error error

+ type ApiVersionsResponseApiKey struct

+ ApiKey int

+ ApiName string

+ MaxVersion int

+ MinVersion int

type Batch

+ func (batch *Batch) Err() error

+ func (batch *Batch) Partition() int

+ type BrokerResolver interface

+ LookupBrokerIPAddr func(ctx context.Context, broker Broker) ([]net.IPAddr, error)

+ func NewBrokerResolver(r *net.Resolver) BrokerResolver

+ type Bytes = protocol.Bytes

+ func NewBytes(b []byte) Bytes

+ type CRC32Balancer struct

+ Consistent bool

+ func (b CRC32Balancer) Balance(msg Message, partitions ...int) (partition int)

+ type Client struct

+ Addr net.Addr

+ Timeout time.Duration

+ Transport RoundTripper

+ func (c *Client) AddOffsetsToTxn(ctx context.Context, req *AddOffsetsToTxnRequest) (*AddOffsetsToTxnResponse, error)

+ func (c *Client) AddPartitionsToTxn(ctx context.Context, req *AddPartitionsToTxnRequest) (*AddPartitionsToTxnResponse, error)

+ func (c *Client) AlterConfigs(ctx context.Context, req *AlterConfigsRequest) (*AlterConfigsResponse, error)

+ func (c *Client) AlterPartitionReassignments(ctx context.Context, req *AlterPartitionReassignmentsRequest) (*AlterPartitionReassignmentsResponse, error)

+ func (c *Client) ApiVersions(ctx context.Context, req *ApiVersionsRequest) (*ApiVersionsResponse, error)

+ func (c *Client) ConsumerOffsets(ctx context.Context, tg TopicAndGroup) (map[int]int64, error)

+ func (c *Client) CreateACLs(ctx context.Context, req *CreateACLsRequest) (*CreateACLsResponse, error)

+ func (c *Client) CreatePartitions(ctx context.Context, req *CreatePartitionsRequest) (*CreatePartitionsResponse, error)

+ func (c *Client) CreateTopics(ctx context.Context, req *CreateTopicsRequest) (*CreateTopicsResponse, error)

+ func (c *Client) DeleteTopics(ctx context.Context, req *DeleteTopicsRequest) (*DeleteTopicsResponse, error)

+ func (c *Client) DescribeConfigs(ctx context.Context, req *DescribeConfigsRequest) (*DescribeConfigsResponse, error)

+ func (c *Client) DescribeGroups(ctx context.Context, req *DescribeGroupsRequest) (*DescribeGroupsResponse, error)

+ func (c *Client) ElectLeaders(ctx context.Context, req *ElectLeadersRequest) (*ElectLeadersResponse, error)

+ func (c *Client) EndTxn(ctx context.Context, req *EndTxnRequest) (*EndTxnResponse, error)

+ func (c *Client) Fetch(ctx context.Context, req *FetchRequest) (*FetchResponse, error)

+ func (c *Client) FindCoordinator(ctx context.Context, req *FindCoordinatorRequest) (*FindCoordinatorResponse, error)

+ func (c *Client) Heartbeat(ctx context.Context, req *HeartbeatRequest) (*HeartbeatResponse, error)

+ func (c *Client) IncrementalAlterConfigs(ctx context.Context, req *IncrementalAlterConfigsRequest) (*IncrementalAlterConfigsResponse, error)

+ func (c *Client) InitProducerID(ctx context.Context, req *InitProducerIDRequest) (*InitProducerIDResponse, error)

+ func (c *Client) ListGroups(ctx context.Context, req *ListGroupsRequest) (*ListGroupsResponse, error)

+ func (c *Client) ListOffsets(ctx context.Context, req *ListOffsetsRequest) (*ListOffsetsResponse, error)

+ func (c *Client) Metadata(ctx context.Context, req *MetadataRequest) (*MetadataResponse, error)

+ func (c *Client) OffsetCommit(ctx context.Context, req *OffsetCommitRequest) (*OffsetCommitResponse, error)

+ func (c *Client) OffsetFetch(ctx context.Context, req *OffsetFetchRequest) (*OffsetFetchResponse, error)

+ func (c *Client) Produce(ctx context.Context, req *ProduceRequest) (*ProduceResponse, error)

+ func (c *Client) TxnOffsetCommit(ctx context.Context, req *TxnOffsetCommitRequest) (*TxnOffsetCommitResponse, error)

+ type Compression = compress.Compression

+ const Gzip

+ const Lz4

+ const Snappy

+ const Zstd

+ type ConfigOperation int8

+ const ConfigOperationAppend

+ const ConfigOperationDelete

+ const ConfigOperationSet

+ const ConfigOperationSubtract

type Conn

+ func (c *Conn) Broker() Broker

type ConnConfig

+ Broker int

+ Rack string

+ type ConsumerGroup struct

+ func NewConsumerGroup(config ConsumerGroupConfig) (*ConsumerGroup, error)

+ func (cg *ConsumerGroup) Close() error

+ func (cg *ConsumerGroup) Next(ctx context.Context) (*Generation, error)

+ type ConsumerGroupConfig struct

+ Brokers []string

+ Dialer *Dialer

+ ErrorLogger Logger

+ GroupBalancers []GroupBalancer

+ HeartbeatInterval time.Duration

+ ID string

+ JoinGroupBackoff time.Duration

+ Logger Logger

+ PartitionWatchInterval time.Duration

+ RebalanceTimeout time.Duration

+ RetentionTime time.Duration

+ SessionTimeout time.Duration

+ StartOffset int64

+ Timeout time.Duration

+ Topics []string

+ WatchPartitionChanges bool

+ func (config *ConsumerGroupConfig) Validate() error

+ type CoordinatorKeyType int8

+ const CoordinatorKeyTypeConsumer

+ const CoordinatorKeyTypeTransaction

+ type CreateACLsRequest struct

+ ACLs []ACLEntry

+ Addr net.Addr

+ type CreateACLsResponse struct

+ Errors []error

+ Throttle time.Duration

+ type CreatePartitionsRequest struct

+ Addr net.Addr

+ Topics []TopicPartitionsConfig

+ ValidateOnly bool

+ type CreatePartitionsResponse struct

+ Errors map[string]error

+ Throttle time.Duration

+ type CreateTopicsRequest struct

+ Addr net.Addr

+ Topics []TopicConfig

+ ValidateOnly bool

+ type CreateTopicsResponse struct

+ Errors map[string]error

+ Throttle time.Duration

+ type DeleteTopicsRequest struct

+ Addr net.Addr

+ Topics []string

+ type DeleteTopicsResponse struct

+ Errors map[string]error

+ Throttle time.Duration

+ type DescribeConfigRequestResource struct

+ ConfigNames []string

+ ResourceName string

+ ResourceType ResourceType

+ type DescribeConfigResponseConfigEntry struct

+ ConfigDocumentation string

+ ConfigName string

+ ConfigSource int8

+ ConfigSynonyms []DescribeConfigResponseConfigSynonym

+ ConfigType int8

+ ConfigValue string

+ IsDefault bool

+ IsSensitive bool

+ ReadOnly bool

+ type DescribeConfigResponseConfigSynonym struct

+ ConfigName string

+ ConfigSource int8

+ ConfigValue string

+ type DescribeConfigResponseResource struct

+ ConfigEntries []DescribeConfigResponseConfigEntry

+ Error error

+ ResourceName string

+ ResourceType int8

+ type DescribeConfigsRequest struct

+ Addr net.Addr

+ IncludeDocumentation bool

+ IncludeSynonyms bool

+ Resources []DescribeConfigRequestResource

+ type DescribeConfigsResponse struct

+ Resources []DescribeConfigResponseResource

+ Throttle time.Duration

+ type DescribeGroupsRequest struct

+ Addr net.Addr

+ GroupIDs []string

+ type DescribeGroupsResponse struct

+ Groups []DescribeGroupsResponseGroup

+ type DescribeGroupsResponseAssignments struct

+ Topics []GroupMemberTopic

+ UserData []byte

+ Version int

+ type DescribeGroupsResponseGroup struct

+ Error error

+ GroupID string

+ GroupState string

+ Members []DescribeGroupsResponseMember

+ type DescribeGroupsResponseMember struct

+ ClientHost string

+ ClientID string

+ MemberAssignments DescribeGroupsResponseAssignments

+ MemberID string

+ MemberMetadata DescribeGroupsResponseMemberMetadata

+ type DescribeGroupsResponseMemberMetadata struct

+ OwnedPartitions []DescribeGroupsResponseMemberMetadataOwnedPartition

+ Topics []string

+ UserData []byte

+ Version int

+ type DescribeGroupsResponseMemberMetadataOwnedPartition struct

+ Partitions []int

+ Topic string

type Dialer

+ DialFunc func(ctx context.Context, network string, address string) (net.Conn, error)

+ type ElectLeadersRequest struct

+ Addr net.Addr

+ Partitions []int

+ Timeout time.Duration

+ Topic string

+ type ElectLeadersResponse struct

+ Error error

+ PartitionResults []ElectLeadersResponsePartitionResult

+ type ElectLeadersResponsePartitionResult struct

+ Error error

+ Partition int

+ type EndTxnRequest struct

+ Addr net.Addr

+ Committed bool

+ ProducerEpoch int

+ ProducerID int

+ TransactionalID string

+ type EndTxnResponse struct

+ Error error

+ Throttle time.Duration

type Error

+ const ElectionNotNeeded

+ const EligibleLeadersNotAvailable

+ const FencedInstanceID

+ const GroupMaxSizeReached

+ const GroupSubscribedToTopic

+ const InvalidRecord

+ const MemberIDRequired

+ const NetworkException

+ const NoReassignmentInProgress

+ const OffsetNotAvailable

+ const PreferredLeaderNotAvailable

+ const StaleBrokerEpoch

+ const UnstableOffsetCommit

+ type FetchRequest struct

+ Addr net.Addr

+ IsolationLevel IsolationLevel

+ MaxBytes int64

+ MaxWait time.Duration

+ MinBytes int64

+ Offset int64

+ Partition int

+ Topic string

+ type FetchResponse struct

+ Error error

+ HighWatermark int64

+ LastStableOffset int64

+ LogStartOffset int64

+ Partition int

+ Records RecordReader

+ Throttle time.Duration

+ Topic string

+ type FindCoordinatorRequest struct

+ Addr net.Addr

+ Key string

+ KeyType CoordinatorKeyType

+ type FindCoordinatorResponse struct

+ Coordinator *FindCoordinatorResponseCoordinator

+ Error error

+ Throttle time.Duration

+ type FindCoordinatorResponseCoordinator struct

+ Host string

+ NodeID int

+ Port int

+ type Generation struct

+ Assignments map[string][]PartitionAssignment

+ GroupID string

+ ID int32

+ MemberID string

+ func (g *Generation) CommitOffsets(offsets map[string]map[int]int64) error

+ func (g *Generation) Start(fn func(ctx context.Context))

+ type GroupMemberTopic struct

+ Partitions []int

+ Topic string

+ type HeartbeatRequest struct

+ Addr net.Addr

+ GenerationID int32

+ GroupID string

+ GroupInstanceID string

+ MemberID string

+ type HeartbeatResponse struct

+ Error error

+ Throttle time.Duration

+ type IncrementalAlterConfigsRequest struct

+ Addr net.Addr

+ Resources []IncrementalAlterConfigsRequestResource

+ ValidateOnly bool

+ type IncrementalAlterConfigsRequestConfig struct

+ ConfigOperation ConfigOperation

+ Name string

+ Value string

+ type IncrementalAlterConfigsRequestResource struct

+ Configs []IncrementalAlterConfigsRequestConfig

+ ResourceName string

+ ResourceType ResourceType

+ type IncrementalAlterConfigsResponse struct

+ Resources []IncrementalAlterConfigsResponseResource

+ type IncrementalAlterConfigsResponseResource struct

+ Error error

+ ResourceName string

+ ResourceType ResourceType

+ type InitProducerIDRequest struct

+ Addr net.Addr

+ ProducerEpoch int

+ ProducerID int

+ TransactionTimeoutMs int

+ TransactionalID string

+ type InitProducerIDResponse struct

+ Error error

+ Producer *ProducerSession

+ Throttle time.Duration

+ type ListGroupsRequest struct

+ Addr net.Addr

+ type ListGroupsResponse struct

+ Error error

+ Groups []ListGroupsResponseGroup

+ type ListGroupsResponseGroup struct

+ Coordinator int

+ GroupID string

+ type ListOffsetsRequest struct

+ Addr net.Addr

+ IsolationLevel IsolationLevel

+ Topics map[string][]OffsetRequest

+ type ListOffsetsResponse struct

+ Throttle time.Duration

+ Topics map[string][]PartitionOffsets

+ type Logger interface

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

+ type LoggerFunc func(string, ...interface{})

+ func (f LoggerFunc) Printf(msg string, args ...interface{})

type Message

+ HighWaterMark int64

+ type MessageTooLargeError struct

+ Message Message

+ Remaining []Message

+ func (e MessageTooLargeError) Error() string

+ type MetadataRequest struct

+ Addr net.Addr

+ Topics []string

+ type MetadataResponse struct

+ Brokers []Broker

+ ClusterID string

+ Controller Broker

+ Throttle time.Duration

+ Topics []Topic

+ type Murmur2Balancer struct

+ Consistent bool

+ func (b Murmur2Balancer) Balance(msg Message, partitions ...int) (partition int)

+ type OffsetCommit struct

+ Metadata string

+ Offset int64

+ Partition int

+ type OffsetCommitPartition struct

+ Error error

+ Partition int

+ type OffsetCommitRequest struct

+ Addr net.Addr

+ GenerationID int

+ GroupID string

+ InstanceID string

+ MemberID string

+ Topics map[string][]OffsetCommit

+ type OffsetCommitResponse struct

+ Throttle time.Duration

+ Topics map[string][]OffsetCommitPartition

+ type OffsetFetchPartition struct

+ CommittedOffset int64

+ Error error

+ Metadata string

+ Partition int

+ type OffsetFetchRequest struct

+ Addr net.Addr

+ GroupID string

+ Topics map[string][]int

+ type OffsetFetchResponse struct

+ Error error

+ Throttle time.Duration

+ Topics map[string][]OffsetFetchPartition

+ type OffsetRequest struct

+ Partition int

+ Timestamp int64

+ func FirstOffsetOf(partition int) OffsetRequest

+ func LastOffsetOf(partition int) OffsetRequest

+ func TimeOffsetOf(partition int, at time.Time) OffsetRequest

type Partition

+ Error error

+ type PartitionAssignment struct

+ ID int

+ Offset int64

+ type PartitionOffsets struct

+ Error error

+ FirstOffset int64

+ LastOffset int64

+ Offsets map[int64]time.Time

+ Partition int

+ type PatternType int8

+ const PatternTypeAny

+ const PatternTypeLiteral

+ const PatternTypeMatch

+ const PatternTypePrefixed

+ const PatternTypeUnknown

+ type ProduceRequest struct

+ Addr net.Addr

+ Compression Compression

+ MessageVersion int

+ Partition int

+ Records RecordReader

+ RequiredAcks RequiredAcks

+ Topic string

+ TransactionalID string

+ type ProduceResponse struct

+ BaseOffset int64

+ Error error

+ LogAppendTime time.Time

+ LogStartOffset int64

+ RecordErrors map[int]error

+ Throttle time.Duration

+ type ProducerSession struct

+ ProducerEpoch int

+ ProducerID int

+ type RackAffinityGroupBalancer struct

+ Rack string

+ func (r RackAffinityGroupBalancer) AssignGroups(members []GroupMember, partitions []Partition) GroupMemberAssignments

+ func (r RackAffinityGroupBalancer) ProtocolName() string

+ func (r RackAffinityGroupBalancer) UserData() ([]byte, error)

type ReadBatchConfig

+ MaxWait time.Duration

type ReaderConfig

+ GroupTopics []string

+ JoinGroupBackoff time.Duration

+ ReadBackoffMax time.Duration

+ ReadBackoffMin time.Duration

+ StartOffset int64

+ type Record = protocol.Record

+ type RecordReader = protocol.RecordReader

+ func NewRecordReader(records ...Record) RecordReader

+ type Request = protocol.Message

+ type RequiredAcks int

+ const RequireAll

+ const RequireNone

+ const RequireOne

+ func (acks *RequiredAcks) UnmarshalText(b []byte) error

+ func (acks RequiredAcks) MarshalText() ([]byte, error)

+ func (acks RequiredAcks) String() string

+ type ResourceType int8

+ const ResourceTypeAny

+ const ResourceTypeBroker

+ const ResourceTypeCluster

+ const ResourceTypeDelegationToken

+ const ResourceTypeGroup

+ const ResourceTypeTopic

+ const ResourceTypeTransactionalID

+ const ResourceTypeUnknown

+ type Response = protocol.Message

+ type RoundTripper interface

+ RoundTrip func(context.Context, net.Addr, Request) (Response, error)

+ var DefaultTransport RoundTripper = &Transport{ ... }

+ type Topic struct

+ Error error

+ Internal bool

+ Name string

+ Partitions []Partition

+ type TopicAndGroup struct

+ GroupId string

+ Topic string

+ type TopicPartitionAssignment struct

+ BrokerIDs []int32

+ type TopicPartitionsConfig struct

+ Count int32

+ Name string

+ TopicPartitionAssignments []TopicPartitionAssignment

+ type Transport struct

+ ClientID string

+ Context context.Context

+ Dial func(context.Context, string, string) (net.Conn, error)

+ DialTimeout time.Duration

+ IdleTimeout time.Duration

+ MetadataTTL time.Duration

+ Resolver BrokerResolver

+ SASL sasl.Mechanism

+ TLS *tls.Config

+ func (t *Transport) CloseIdleConnections()

+ func (t *Transport) RoundTrip(ctx context.Context, addr net.Addr, req Request) (Response, error)

+ type TxnOffsetCommit struct

+ Metadata string

+ Offset int64

+ Partition int

+ type TxnOffsetCommitPartition struct

+ Error error

+ Partition int

+ type TxnOffsetCommitRequest struct

+ Addr net.Addr

+ GenerationID int

+ GroupID string

+ GroupInstanceID string

+ MemberID string

+ ProducerEpoch int

+ ProducerID int

+ Topics map[string][]TxnOffsetCommit

+ TransactionalID string

+ type TxnOffsetCommitResponse struct

+ Throttle time.Duration

+ Topics map[string][]TxnOffsetCommitPartition

+ type Version int16

+ func (n Version) Marshal(v interface{}) ([]byte, error)

+ func (n Version) Unmarshal(b []byte, v interface{}) error

+ type WriteErrors []error

+ func (err WriteErrors) Count() int

+ func (err WriteErrors) Error() string

type Writer

+ Addr net.Addr

+ Async bool

+ Balancer Balancer

+ BatchBytes int64

+ BatchSize int

+ BatchTimeout time.Duration

+ Completion func(messages []Message, err error)

+ Compression Compression

+ ErrorLogger Logger

+ Logger Logger

+ MaxAttempts int

+ ReadTimeout time.Duration

+ RequiredAcks RequiredAcks

+ Topic string

+ Transport RoundTripper

+ WriteTimeout time.Duration

type WriterConfig

+ BatchBytes int

+ IdleConnTimeout time.Duration

type WriterStats

+ BatchBytes SummaryStats

+ BatchTime DurationStats

v0.2.3

May 3, 2019

Changes in this version

+ type ApiVersion struct

+ ApiKey int16

+ MaxVersion int16

+ MinVersion int16

type Conn

+ func (c *Conn) ApiVersions() ([]ApiVersion, error)

+ func (c *Conn) Brokers() ([]Broker, error)

+ func (c *Conn) Controller() (broker Broker, err error)

+ func (c *Conn) ReadBatchWith(cfg ReadBatchConfig) *Batch

+ func (c *Conn) WriteCompressedMessagesAt(codec CompressionCodec, msgs ...Message) (nbytes int, partition int32, offset int64, appendTime time.Time, err error)

type ConnConfig

+ TransactionalID string

type Dialer

+ SASLMechanism sasl.Mechanism

+ TransactionalID string

type Error

+ const DelegationTokenAuthDisabled

+ const DelegationTokenAuthorizationFailed

+ const DelegationTokenExpired

+ const DelegationTokenNotFound

+ const DelegationTokenOwnerMismatch

+ const DelegationTokenRequestNotAllowed

+ const FencedLeaderEpoch

+ const FetchSessionIDNotFound

+ const GroupIdNotFound

+ const InvalidFetchSessionEpoch

+ const InvalidPrincipalType

+ const KafkaStorageError

+ const ListenerNotFound

+ const LogDirNotFound

+ const NonEmptyGroup

+ const ReassignmentInProgress

+ const SASLAuthenticationFailed

+ const TopicDeletionDisabled

+ const UnknownLeaderEpoch

+ const UnknownProducerId

+ const UnsupportedCompressionType

+ type Header struct

+ Key string

+ Value []byte

+ type IsolationLevel int8

+ const ReadCommitted

+ const ReadUncommitted

type Message

+ Headers []Header

+ type ReadBatchConfig struct

+ IsolationLevel IsolationLevel

+ MaxBytes int

+ MinBytes int

type Reader

+ func (r *Reader) SetOffsetAt(ctx context.Context, t time.Time) error

type ReaderConfig

+ IsolationLevel IsolationLevel

+ MaxAttempts int

+ PartitionWatchInterval time.Duration

+ WatchPartitionChanges bool

+ func (config *ReaderConfig) Validate() error

type WriterConfig

+ func (config *WriterConfig) Validate() error

v0.2.2

Nov 27, 2018

Changes in this version

type ReaderStats

+ DeprecatedFetchesWithTypo int64

v0.2.1

Nov 15, 2018

Changes in this version

type Conn

+ func DialPartition(ctx context.Context, network string, address string, partition Partition) (*Conn, error)

+ func (c *Conn) WriteCompressedMessages(codec CompressionCodec, msgs ...Message) (int, error)

type Dialer

+ func (d *Dialer) DialPartition(ctx context.Context, network string, address string, partition Partition) (*Conn, error)

+ func (d *Dialer) LookupPartition(ctx context.Context, network string, address string, topic string, ...) (Partition, error)

type Partition

+ func LookupPartition(ctx context.Context, network string, address string, topic string, ...) (Partition, error)

+ func LookupPartitions(ctx context.Context, network string, address string, topic string) ([]Partition, error)

type WriterConfig

+ ErrorLogger *log.Logger

+ Logger *log.Logger

v0.2.0

Oct 25, 2018

Changes in this version

+ const FirstOffset

+ const LastOffset

type Broker

+ Rack string

+ type GroupBalancer interface

+ AssignGroups func(members []GroupMember, partitions []Partition) GroupMemberAssignments

+ ProtocolName func() string

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

+ type GroupMember struct

+ ID string

+ Topics []string

+ UserData []byte

+ type GroupMemberAssignments map[string]map[string][]int

+ type RangeGroupBalancer struct

+ func (r RangeGroupBalancer) AssignGroups(members []GroupMember, topicPartitions []Partition) GroupMemberAssignments

+ func (r RangeGroupBalancer) ProtocolName() string

+ func (r RangeGroupBalancer) UserData() ([]byte, error)

type ReaderConfig

+ GroupBalancers []GroupBalancer

+ type RoundRobinGroupBalancer struct

+ func (r RoundRobinGroupBalancer) AssignGroups(members []GroupMember, topicPartitions []Partition) GroupMemberAssignments

+ func (r RoundRobinGroupBalancer) ProtocolName() string

+ func (r RoundRobinGroupBalancer) UserData() ([]byte, error)

v0.1.0

Jul 16, 2018

Changes in this version

+ const CompressionNoneCode

+ const DefaultCompressionLevel

+ const SeekAbsolute

+ const SeekCurrent

+ const SeekEnd

+ const SeekStart

+ var DefaultClientID string

+ var DefaultDialer = &Dialer

+ func RegisterCompressionCodec(codec func() CompressionCodec)

+ type Balancer interface

+ Balance func(msg Message, partitions ...int) (partition int)

+ type BalancerFunc func(Message, ...int) int

+ func (f BalancerFunc) Balance(msg Message, partitions ...int) int

+ type Batch struct

+ func (batch *Batch) Close() error

+ func (batch *Batch) HighWaterMark() int64

+ func (batch *Batch) Offset() int64

+ func (batch *Batch) Read(b []byte) (int, error)

+ func (batch *Batch) ReadMessage() (Message, error)

+ func (batch *Batch) Throttle() time.Duration

+ type Broker struct

+ Host string

+ ID int

+ Port int

+ type CompressionCodec interface

+ Code func() int8

+ Decode func(dst, src []byte) (int, error)

+ Encode func(dst, src []byte) (int, error)

+ type ConfigEntry struct

+ ConfigName string

+ ConfigValue string

+ type Conn struct

+ func Dial(network string, address string) (*Conn, error)

+ func DialContext(ctx context.Context, network string, address string) (*Conn, error)

+ func DialLeader(ctx context.Context, network string, address string, topic string, ...) (*Conn, error)

+ func NewConn(conn net.Conn, topic string, partition int) *Conn

+ func NewConnWith(conn net.Conn, config ConnConfig) *Conn

+ func (c *Conn) Close() error

+ func (c *Conn) CreateTopics(topics ...TopicConfig) error

+ func (c *Conn) DeleteTopics(topics ...string) error

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

+ func (c *Conn) Offset() (offset int64, whence int)

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

+ func (c *Conn) ReadBatch(minBytes int, maxBytes int) *Batch

+ func (c *Conn) ReadFirstOffset() (int64, error)

+ func (c *Conn) ReadLastOffset() (int64, error)

+ func (c *Conn) ReadMessage(maxBytes int) (Message, error)

+ func (c *Conn) ReadOffset(t time.Time) (int64, error)

+ func (c *Conn) ReadOffsets() (first int64, last int64, err error)

+ func (c *Conn) ReadPartitions(topics ...string) (partitions []Partition, err error)

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

+ func (c *Conn) Seek(offset int64, whence int) (int64, error)

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

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

+ func (c *Conn) SetRequiredAcks(n int) error

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

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

+ func (c *Conn) WriteMessages(msgs ...Message) (int, error)

+ type ConnConfig struct

+ ClientID string

+ Partition int

+ Topic string

+ type Dialer struct

+ ClientID string

+ Deadline time.Time

+ DualStack bool

+ FallbackDelay time.Duration

+ KeepAlive time.Duration

+ LocalAddr net.Addr

+ Resolver Resolver

+ TLS *tls.Config

+ Timeout time.Duration

+ func (d *Dialer) Dial(network string, address string) (*Conn, error)

+ func (d *Dialer) DialContext(ctx context.Context, network string, address string) (*Conn, error)

+ func (d *Dialer) DialLeader(ctx context.Context, network string, address string, topic string, ...) (*Conn, error)

+ func (d *Dialer) LookupLeader(ctx context.Context, network string, address string, topic string, ...) (Broker, error)

+ func (d *Dialer) LookupPartitions(ctx context.Context, network string, address string, topic string) ([]Partition, error)

+ type DurationStats struct

+ Avg time.Duration

+ Max time.Duration

+ Min time.Duration

+ type Error int

+ const BrokerAuthorizationFailed

+ const BrokerNotAvailable

+ const ClusterAuthorizationFailed

+ const ConcurrentTransactions

+ const DuplicateSequenceNumber

+ const GroupAuthorizationFailed

+ const GroupCoordinatorNotAvailable

+ const GroupLoadInProgress

+ const IllegalGeneration

+ const IllegalSASLState

+ const InconsistentGroupProtocol

+ const InvalidCommitOffsetSize

+ const InvalidConfiguration

+ const InvalidGroupId

+ const InvalidMessage

+ const InvalidMessageSize

+ const InvalidPartitionNumber

+ const InvalidProducerEpoch

+ const InvalidProducerIDMapping

+ const InvalidReplicaAssignment

+ const InvalidReplicationFactor

+ const InvalidRequest

+ const InvalidRequiredAcks

+ const InvalidSessionTimeout

+ const InvalidTimestamp

+ const InvalidTopic

+ const InvalidTransactionState

+ const InvalidTransactionTimeout

+ const LeaderNotAvailable

+ const MessageSizeTooLarge

+ const NotController

+ const NotCoordinatorForGroup

+ const NotEnoughReplicas

+ const NotEnoughReplicasAfterAppend

+ const NotLeaderForPartition

+ const OffsetMetadataTooLarge

+ const OffsetOutOfRange

+ const OutOfOrderSequenceNumber

+ const PolicyViolation

+ const RebalanceInProgress

+ const RecordListTooLarge

+ const ReplicaNotAvailable

+ const RequestTimedOut

+ const SecurityDisabled

+ const StaleControllerEpoch

+ const TopicAlreadyExists

+ const TopicAuthorizationFailed

+ const TransactionCoordinatorFenced

+ const TransactionalIDAuthorizationFailed

+ const Unknown

+ const UnknownMemberId

+ const UnknownTopicOrPartition

+ const UnsupportedForMessageFormat

+ const UnsupportedSASLMechanism

+ const UnsupportedVersion

+ func (e Error) Description() string

+ func (e Error) Error() string

+ func (e Error) Temporary() bool

+ func (e Error) Timeout() bool

+ func (e Error) Title() string

+ type Hash struct

+ Hasher hash.Hash32

+ func (h *Hash) Balance(msg Message, partitions ...int) (partition int)

+ type LeastBytes struct

+ func (lb *LeastBytes) Balance(msg Message, partitions ...int) int

+ type ListGroupsResponseGroupV1 struct

+ GroupID string

+ ProtocolType string

+ type Message struct

+ Key []byte

+ Offset int64

+ Partition int

+ Time time.Time

+ Topic string

+ Value []byte

+ type Partition struct

+ ID int

+ Isr []Broker

+ Leader Broker

+ Replicas []Broker

+ Topic string

+ type Reader struct

+ func NewReader(config ReaderConfig) *Reader

+ func (r *Reader) Close() error

+ func (r *Reader) CommitMessages(ctx context.Context, msgs ...Message) error

+ func (r *Reader) Config() ReaderConfig

+ func (r *Reader) FetchMessage(ctx context.Context) (Message, error)

+ func (r *Reader) Lag() int64

+ func (r *Reader) Offset() int64

+ func (r *Reader) ReadLag(ctx context.Context) (lag int64, err error)

+ func (r *Reader) ReadMessage(ctx context.Context) (Message, error)

+ func (r *Reader) SetOffset(offset int64) error

+ func (r *Reader) Stats() ReaderStats

+ type ReaderConfig struct

+ Brokers []string

+ CommitInterval time.Duration

+ Dialer *Dialer

+ ErrorLogger *log.Logger

+ GroupID string

+ HeartbeatInterval time.Duration

+ Logger *log.Logger

+ MaxBytes int

+ MaxWait time.Duration

+ MinBytes int

+ Partition int

+ QueueCapacity int

+ ReadLagInterval time.Duration

+ RebalanceTimeout time.Duration

+ RetentionTime time.Duration

+ SessionTimeout time.Duration

+ Topic string

+ type ReaderStats struct

+ Bytes int64

+ ClientID string

+ DialTime DurationStats

+ Dials int64

+ Errors int64

+ FetchBytes SummaryStats

+ FetchSize SummaryStats

+ Fetches int64

+ Lag int64

+ MaxBytes int64

+ MaxWait time.Duration

+ Messages int64

+ MinBytes int64

+ Offset int64

+ Partition string

+ QueueCapacity int64

+ QueueLength int64

+ ReadTime DurationStats

+ Rebalances int64

+ Timeouts int64

+ Topic string

+ WaitTime DurationStats

+ type ReplicaAssignment struct

+ Partition int

+ Replicas int

+ type Resolver interface

+ LookupHost func(ctx context.Context, host string) (addrs []string, err error)

+ type RoundRobin struct

+ func (rr *RoundRobin) Balance(msg Message, partitions ...int) int

+ type SummaryStats struct

+ Avg int64

+ Max int64

+ Min int64

+ type TopicConfig struct

+ ConfigEntries []ConfigEntry

+ NumPartitions int

+ ReplicaAssignments []ReplicaAssignment

+ ReplicationFactor int

+ Topic string

+ type Writer struct

+ func NewWriter(config WriterConfig) *Writer

+ func (w *Writer) Close() (err error)

+ func (w *Writer) Stats() WriterStats

+ func (w *Writer) WriteMessages(ctx context.Context, msgs ...Message) error

+ type WriterConfig struct

+ Async bool

+ Balancer Balancer

+ BatchSize int

+ BatchTimeout time.Duration

+ Brokers []string

+ Dialer *Dialer

+ MaxAttempts int

+ QueueCapacity int

+ ReadTimeout time.Duration

+ RebalanceInterval time.Duration

+ RequiredAcks int

+ Topic string

+ WriteTimeout time.Duration

+ type WriterStats struct

+ Async bool

+ BatchSize SummaryStats

+ BatchTimeout time.Duration

+ Bytes int64

+ ClientID string

+ DialTime DurationStats

+ Dials int64

+ Errors int64

+ MaxAttempts int64

+ MaxBatchSize int64

+ Messages int64

+ QueueCapacity int64

+ QueueLength int64

+ ReadTimeout time.Duration

+ RebalanceInterval time.Duration

+ Rebalances int64

+ RequiredAcks int64

+ Retries SummaryStats

+ Topic string

+ WaitTime DurationStats

+ WriteTime DurationStats

+ WriteTimeout time.Duration

+ Writes int64

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