kgo

Oct 15, 2024

Changes in this version

type ErrGroupSession

+ Err error

type FetchTopic

+ TopicID [16]byte

type GroupTransactSession

+ func (s *GroupTransactSession) AllowRebalance()

+ func (s *GroupTransactSession) CloseAllowingRebalance()

May 26, 2024

Changes in this version

type Client

+ func (cl *Client) GetConsumeTopics() []string

+ func (cl *Client) MarkCommitOffsets(unmarked map[string]map[int32]EpochOffset)

type Offset

+ func (o Offset) AtCommitted() Offset

Jan 21, 2024

Changes in this version

type PartitionerHasher

+ func SaramaCompatHasher(hashFn func([]byte) uint32) PartitionerHasher

Sep 21, 2023

Changes in this version

+ func PreTxnCommitFnContext(ctx context.Context, fn func(*kmsg.TxnOffsetCommitRequest) error) context.Context

type Client

+ func (cl *Client) BufferedFetchBytes() int64

+ func (cl *Client) BufferedProduceBytes() int64

+ func (cl *Client) LeaveGroupContext(ctx context.Context) error

+ func (cl *Client) PurgeTopicsFromConsuming(topics ...string)

+ func (cl *Client) PurgeTopicsFromProducing(topics ...string)

type Fetches

+ func NewErrFetch(err error) Fetches

type Opt

+ func ConsiderMissingTopicDeletedAfter(t time.Duration) Opt

type ProducerOpt

+ func MaxBufferedBytes(n int) ProducerOpt

Jul 8, 2023

Changes in this version

+ func NodeName(nodeID int32) string

type Client

+ func (cl *Client) AddConsumePartitions(partitions map[string]map[int32]Offset)

+ func (cl *Client) Opts() []Opt

+ func (cl *Client) RemoveConsumePartitions(partitions map[string][]int32)

type ConsumerOpt

+ func KeepRetryableFetchErrors() ConsumerOpt

type Offset

+ func (o Offset) EpochOffset() EpochOffset

type Opt

+ func DialTLS() Opt

Mar 14, 2023

Changes in this version

type Client

+ func (cl *Client) OptValue(opt any) any

+ func (cl *Client) OptValues(opt any) []any

+ func (cl *Client) UpdateFetchMaxBytes(maxBytes, maxPartBytes int32)

+ type ErrGroupSession struct

+ func (e *ErrGroupSession) Error() string

+ func (e *ErrGroupSession) Unwrap() error

+ type HookClientClosed interface

+ OnClientClosed func(*Client)

+ type HookProduceRecordPartitioned interface

+ OnProduceRecordPartitioned func(*Record, int32)

Feb 8, 2023

Changes in this version

type Client

+ func (cl *Client) CommitMarkedOffsets(ctx context.Context) error

+ func (cl *Client) UpdateSeedBrokers(addrs ...string) error

Jan 4, 2023

Changes in this version

type Client

+ func (cl *Client) MarkedOffsets() map[string]map[int32]EpochOffset

+ func (cl *Client) UnsafeAbortBufferedRecords()

Nov 17, 2022

Changes in this version

type ConsumerOpt

+ func ConsumePreferringLagFn(fn PreferLagFn) ConsumerOpt

+ type PreferLagFn func(lag map[string]map[int32]int64, torderPrior []string, ...) ([]string, map[string][]int32)

+ func PreferLagAt(preferLagAt int64) PreferLagFn

Oct 20, 2022

Changes in this version

+ type ErrFirstReadEOF struct

+ func (*ErrFirstReadEOF) Error() string

+ func (*ErrFirstReadEOF) Unwrap() error

type Opt

+ func DialTimeout(timeout time.Duration) Opt

Oct 7, 2022

Changes in this version

+ func ValidateOpts(opts ...Opt) error

type Record

+ Context context.Context

Aug 22, 2022

Changes in this version

type Client

+ func (cl *Client) CloseAllowingRebalance()

+ func (cl *Client) GroupMetadata() (string, int32)

+ func (cl *Client) PartitionLeader(topic string, partition int32) (leader, leaderEpoch int32, err error)

+ func (cl *Client) ProducerID(ctx context.Context) (int64, int16, error)

type EpochOffset

+ func (e EpochOffset) Less(o EpochOffset) bool

type Fetches

+ func (fs Fetches) Empty() bool

+ func (fs Fetches) Err0() error

+ func (fs Fetches) NumRecords() (n int)

Jun 1, 2022

Changes in this version

type Partitioner

+ func UniformBytesPartitioner(bytes int, adaptive, keys bool, hasher PartitionerHasher) Partitioner

May 5, 2022

Changes in this version

type Fetches

+ func (fs Fetches) Err() error

May 2, 2022

Changes in this version

+ func PreCommitFnContext(ctx context.Context, fn func(*kmsg.OffsetCommitRequest) error) context.Context

type BalancePlan

+ func (p *BalancePlan) AsMemberIDMap() map[string]map[string][]int32

type ConsumerBalancer

+ func (b *ConsumerBalancer) BalanceOrError(topics map[string]int32) (IntoSyncAssignment, error)

+ func (b *ConsumerBalancer) SetError(err error)

+ type GroupMemberBalancerOrError interface

+ BalanceOrError func(topics map[string]int32) (IntoSyncAssignment, error)

type GroupOpt

+ func OnOffsetsFetched(onFetched func(context.Context, *Client, *kmsg.OffsetFetchResponse) error) GroupOpt

type Offset

+ func (o Offset) AfterMilli(millisec int64) Offset

Mar 1, 2022

Changes in this version

type Client

+ func (cl *Client) AddConsumeTopics(topics ...string)

+ func (cl *Client) AllowRebalance()

+ func (cl *Client) EndAndBeginTransaction(ctx context.Context, how EndBeginTxnHow, commit TransactionEndTry, ...) (rerr error)

+ func (cl *Client) ForceMetadataRefresh()

+ func (cl *Client) PurgeTopicsFromClient(topics ...string)

+ func (cl *Client) TryProduce(ctx context.Context, r *Record, promise func(*Record, error))

+ type EndBeginTxnHow uint8

+ const EndBeginTxnSafe

+ const EndBeginTxnUnsafe

type GroupOpt

+ func BlockRebalanceOnPoll() GroupOpt

type GroupTransactSession

+ func (s *GroupTransactSession) TryProduce(ctx context.Context, r *Record, promise func(*Record, error))

type Offset

+ func NoResetOffset() Offset

type Opt

+ func ConcurrentTransactionsBackoff(backoff time.Duration) Opt

type ProducerOpt

+ func MaxProduceRequestsInflightPerBroker(n int) ProducerOpt

+ func UnknownTopicRetries(n int) ProducerOpt

Jan 6, 2022

Changes in this version

type Client

+ func (cl *Client) Ping(ctx context.Context) error

type GroupOpt

+ func AdjustFetchOffsetsFn(...) GroupOpt

Nov 12, 2021

Changes in this version

type ConsumerOpt

+ func DisableFetchSessions() ConsumerOpt

Nov 5, 2021

Changes in this version

type FetchTopic

+ func (t *FetchTopic) EachRecord(fn func(*Record))

+ func (t *FetchTopic) Records() []*Record

type Offset

+ func (o Offset) String() string

Oct 21, 2021

Changes in this version

type Record

+ func (r *Record) AppendFormat(b []byte, layout string) ([]byte, error)

+ type RecordFormatter struct

+ func NewRecordFormatter(layout string) (*RecordFormatter, error)

+ func (f *RecordFormatter) AppendPartitionRecord(b []byte, p *FetchPartition, r *Record) []byte

+ func (f *RecordFormatter) AppendRecord(b []byte, r *Record) []byte

+ type RecordReader struct

+ func NewRecordReader(reader io.Reader, layout string) (*RecordReader, error)

+ func (r *RecordReader) ReadRecord() (*Record, error)

+ func (r *RecordReader) ReadRecordInto(rec *Record) error

+ func (r *RecordReader) SetReader(reader io.Reader)

Sep 21, 2021

Changes in this version

+ var ErrRecordRetries = errors.New("record failed after being retried too many times")

type FetchPartition

+ func (p *FetchPartition) EachRecord(fn func(*Record))

type FetchTopic

+ func (t *FetchTopic) EachPartition(fn func(FetchPartition))

Sep 3, 2021

Changes in this version

type Client

+ func (cl *Client) PauseFetchPartitions(topicPartitions map[string][]int32) map[string][]int32

+ func (cl *Client) PauseFetchTopics(topics ...string) []string

+ func (cl *Client) ResumeFetchPartitions(topicPartitions map[string][]int32)

+ func (cl *Client) ResumeFetchTopics(topics ...string)

type Fetches

+ func (fs Fetches) Records() []*Record

Sep 1, 2021

Changes in this version

type GroupOpt

+ func AutoCommitCallback(fn func(*Client, *kmsg.OffsetCommitRequest, *kmsg.OffsetCommitResponse, error)) GroupOpt

+ func OnPartitionsAssigned(onAssigned func(context.Context, *Client, map[string][]int32)) GroupOpt

+ func OnPartitionsLost(onLost func(context.Context, *Client, map[string][]int32)) GroupOpt

+ func OnPartitionsRevoked(onRevoked func(context.Context, *Client, map[string][]int32)) GroupOpt

type Opt

+ func RequestTimeoutOverhead(overhead time.Duration) Opt

type ProducerOpt

+ func ProducerBatchCompression(preference ...CompressionCodec) ProducerOpt

+ func ProducerBatchMaxBytes(v int32) ProducerOpt

+ func ProducerLinger(linger time.Duration) ProducerOpt

+ func ProducerOnDataLossDetected(fn func(string, int32)) ProducerOpt

+ func RecordDeliveryTimeout(timeout time.Duration) ProducerOpt

+ func StopProducerOnDataLossDetected() ProducerOpt

Sep 1, 2021

Changes in this version

type Client

+ func (cl *Client) MarkCommitRecords(rs ...*Record)

type ConsumerOpt

+ func MaxConcurrentFetches(n int) ConsumerOpt

type Fetches

+ func (fs Fetches) CollectRecords() []*Record

type GroupOpt

+ func AutoCommitMarks() GroupOpt

type Offset

+ func (o Offset) MarshalJSON() ([]byte, error)

Aug 25, 2021

Changes in this version

type GroupOpt

+ func GreedyAutoCommit() GroupOpt

type Partitioner

+ func RoundRobinPartitioner() Partitioner

+ type TopicBackupIter interface

+ Next func() (int, int64)

+ Rem func() int

+ type TopicPartitionerOnNewBatch interface

+ OnNewBatch func()

Aug 20, 2021

Changes in this version

type Partitioner

+ func LeastBackupPartitioner() Partitioner

+ type TopicBackupPartitioner interface

+ PartitionByBackup func(r *Record, n int, backupFn func() (int, int64)) int

Aug 17, 2021

Changes in this version

type Opt

+ func DialTLSConfig(c *tls.Config) Opt

+ func RetryBackoffFn(backoff func(int) time.Duration) Opt

+ func RetryTimeoutFn(t func(int16) time.Duration) Opt

type ProducerOpt

+ func RecordRetries(n int) ProducerOpt

Jul 15, 2021

Changes in this version

type Client

+ func (cl *Client) BufferedFetchRecords() int64

+ func (cl *Client) BufferedProduceRecords() int64

+ type HookFetchRecordBuffered interface

+ OnFetchRecordBuffered func(*Record)

+ type HookFetchRecordUnbuffered interface

+ OnFetchRecordUnbuffered func(r *Record, polled bool)

+ type HookNewClient interface

+ OnNewClient func(*Client)

+ type HookProduceRecordBuffered interface

+ OnProduceRecordBuffered func(*Record)

+ type HookProduceRecordUnbuffered interface

+ OnProduceRecordUnbuffered func(*Record, error)

Jun 13, 2021

Changes in this version

+ var ErrClientClosed = errors.New("client closed")

type Client

+ func (cl *Client) CommitOffsetsSync(ctx context.Context, uncommitted map[string]map[int32]EpochOffset, ...)

+ func (cl *Client) CommitRecords(ctx context.Context, rs ...*Record) error

+ func (cl *Client) CommitUncommittedOffsets(ctx context.Context) error

+ type FetchBatchMetrics struct

+ CompressedBytes int

+ CompressionType uint8

+ NumRecords int

+ UncompressedBytes int

type Fetches

+ func (fs Fetches) IsClientClosed() bool

type HookBrokerThrottle

+ OnBrokerThrottle func(meta BrokerMetadata, throttleInterval time.Duration, ...)

+ type HookFetchBatchRead interface

+ OnFetchBatchRead func(meta BrokerMetadata, topic string, partition int32, metrics FetchBatchMetrics)

+ type HookProduceBatchWritten interface

+ OnProduceBatchWritten func(meta BrokerMetadata, topic string, partition int32, ...)

type Partitioner

+ func BasicConsistentPartitioner(partition func(string) func(r *Record, n int) int) Partitioner

+ func ManualPartitioner() Partitioner

+ type ProduceBatchMetrics struct

+ CompressedBytes int

+ CompressionType uint8

+ NumRecords int

+ UncompressedBytes int

Jun 7, 2021

Changes in this version

+ type BrokerE2E struct

+ BytesRead int

+ BytesWritten int

+ ReadErr error

+ ReadWait time.Duration

+ TimeToRead time.Duration

+ TimeToWrite time.Duration

+ WriteErr error

+ WriteWait time.Duration

+ func (e *BrokerE2E) DurationE2E() time.Duration

+ func (e *BrokerE2E) Err() error

type ConsumerOpt

+ func ConsumeRegex() ConsumerOpt

type Fetches

+ func (fs Fetches) EachError(fn func(string, int32, error))

type GroupOpt

+ func ConsumerGroup(group string) GroupOpt

type GroupTransactSession

+ func NewGroupTransactSession(opts ...Opt) (*GroupTransactSession, error)

+ func (s *GroupTransactSession) Client() *Client

+ func (s *GroupTransactSession) Close()

type HookBrokerConnect

+ OnBrokerConnect func(meta BrokerMetadata, dialDur time.Duration, conn net.Conn, err error)

type HookBrokerDisconnect

+ OnBrokerDisconnect func(meta BrokerMetadata, conn net.Conn)

type HookBrokerE2E

+ OnBrokerE2E func(meta BrokerMetadata, key int16, e2e BrokerE2E)

type HookBrokerRead

+ OnBrokerRead func(meta BrokerMetadata, key int16, bytesRead int, ...)

type HookBrokerWrite

+ OnBrokerWrite func(meta BrokerMetadata, key int16, bytesWritten int, ...)

type Opt

+ func AllowAutoTopicCreation() Opt

type ProducerOpt

+ func DefaultProduceTopic(t string) ProducerOpt

Jun 2, 2021

Changes in this version

+ type E2EInfo struct

+ BytesRead int

+ BytesWritten int

+ ReadErr error

+ ReadWait time.Duration

+ TimeToRead time.Duration

+ TimeToWrite time.Duration

+ WriteErr error

+ WriteWait time.Duration

+ func (e *E2EInfo) DurationE2E() time.Duration

+ func (e *E2EInfo) Err() error

+ type HookBrokerE2E interface

+ OnE2E func(meta BrokerMetadata, key int16, e2e E2EInfo)

May 21, 2021

Changes in this version

type FirstErrPromise

+ func AbortingFirstErrPromise(cl *Client) *FirstErrPromise

May 13, 2021

Changes in this version

+ func ParseConsumerSyncAssignment(assignment []byte) (map[string][]int32, error)

+ type BalancePlan struct

+ func (p *BalancePlan) AddPartition(member *kmsg.JoinGroupResponseMember, topic string, partition int32)

+ func (p *BalancePlan) AddPartitions(member *kmsg.JoinGroupResponseMember, topic string, partitions []int32)

+ func (p *BalancePlan) AdjustCooperative(b *ConsumerBalancer)

+ func (p *BalancePlan) IntoSyncAssignment() []kmsg.SyncGroupRequestGroupAssignment

+ func (p *BalancePlan) String() string

type Client

+ func (cl *Client) ForceRebalance()

+ func (cl *Client) ProduceSync(ctx context.Context, rs ...*Record) ProduceResults

+ type ConsumerBalancer struct

+ func NewConsumerBalancer(balance ConsumerBalancerBalance, members []kmsg.JoinGroupResponseMember) (*ConsumerBalancer, error)

+ func (b *ConsumerBalancer) Balance(topics map[string]int32) IntoSyncAssignment

+ func (b *ConsumerBalancer) EachMember(fn func(member *kmsg.JoinGroupResponseMember, meta *kmsg.GroupMemberMetadata))

+ func (b *ConsumerBalancer) MemberAt(n int) (*kmsg.JoinGroupResponseMember, *kmsg.GroupMemberMetadata)

+ func (b *ConsumerBalancer) MemberTopics() map[string]struct{}

+ func (b *ConsumerBalancer) Members() []kmsg.JoinGroupResponseMember

+ func (b *ConsumerBalancer) NewPlan() *BalancePlan

+ type ConsumerBalancerBalance interface

+ Balance func(*ConsumerBalancer, map[string]int32) IntoSyncAssignment

type Fetches

+ func (fs Fetches) EachRecord(fn func(*Record))

+ func (fs Fetches) EachTopic(fn func(FetchTopic))

+ type FirstErrPromise struct

+ func (f *FirstErrPromise) Err() error

+ func (f *FirstErrPromise) Promise(_ *Record, err error)

+ func (f *FirstErrPromise) PromiseFn() func(*Record, error)

type GroupBalancer

+ IsCooperative func() bool

+ JoinGroupMetadata func(topicInterests []string, currentAssignment map[string][]int32, ...) []byte

+ MemberBalancer func(members []kmsg.JoinGroupResponseMember) (b GroupMemberBalancer, topics map[string]struct{}, err error)

+ ParseSyncAssignment func(assignment []byte) (map[string][]int32, error)

+ ProtocolName func() string

+ type GroupMemberBalancer interface

+ Balance func(topics map[string]int32) IntoSyncAssignment

type GroupOpt

+ func CommitCallback(fn func(*kmsg.OffsetCommitRequest, *kmsg.OffsetCommitResponse, error)) GroupOpt

type GroupTransactSession

+ func (s *GroupTransactSession) ProduceSync(ctx context.Context, rs ...*Record) ProduceResults

+ type HookBrokerConnect interface

+ OnConnect func(meta BrokerMetadata, dialDur time.Duration, conn net.Conn, err error)

+ type HookBrokerDisconnect interface

+ OnDisconnect func(meta BrokerMetadata, conn net.Conn)

+ type HookBrokerRead interface

+ OnRead func(meta BrokerMetadata, key int16, bytesRead int, ...)

+ type HookBrokerThrottle interface

+ OnThrottle func(meta BrokerMetadata, throttleInterval time.Duration, ...)

+ type HookBrokerWrite interface

+ OnWrite func(meta BrokerMetadata, key int16, bytesWritten int, ...)

+ type HookGroupManageError interface

+ OnGroupManageError func(error)

+ type IntoSyncAssignment interface

+ IntoSyncAssignment func() []kmsg.SyncGroupRequestGroupAssignment

+ type ProduceResult struct

+ Err error

+ Record *Record

+ type ProduceResults []ProduceResult

+ func (rs ProduceResults) First() (*Record, error)

+ func (rs ProduceResults) FirstErr() error

type ProducerOpt

+ func ProduceTopic(t string) ProducerOpt

type Record

+ func KeySliceRecord(key, value []byte) *Record

+ func KeyStringRecord(key, value string) *Record

+ func SliceRecord(value []byte) *Record

+ func StringRecord(value string) *Record

Apr 20, 2021

Changes in this version

type Client

+ func (cl *Client) ResetProducerID() bool

type Opt

+ func ConnIdleTimeout(timeout time.Duration) Opt

+ func ProduceRetries(n int) Opt

Mar 29, 2021

Changes in this version

type Client

+ func (cl *Client) PollRecords(ctx context.Context, maxPollRecords int) Fetches

+ type FetchTopicPartition struct

+ Partition FetchPartition

+ Topic string

+ func (r *FetchTopicPartition) EachRecord(fn func(*Record))

type Fetches

+ func (fs Fetches) EachPartition(fn func(FetchTopicPartition))

type GroupOpt

+ func GroupProtocol(protocol string) GroupOpt

type GroupTransactSession

+ func (s *GroupTransactSession) PollFetches(ctx context.Context) Fetches

+ func (s *GroupTransactSession) PollRecords(ctx context.Context, maxPollRecords int) Fetches

+ func (s *GroupTransactSession) Produce(ctx context.Context, r *Record, promise func(*Record, error)) error

type ProducerOpt

+ func DisableIdempotentWrite() ProducerOpt

Feb 10, 2021

Changes in this version

type Client

+ func (cl *Client) LeaveGroup()

type ConsumerOpt

+ func AllowedConcurrentFetches(n int) ConsumerOpt

Jan 13, 2021

Changes in this version

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

+ var ErrInvalidPartition = errors.New("invalid partition chosen from partitioner")

type Broker

+ func (b *Broker) RetriableRequest(ctx context.Context, req kmsg.Request) (kmsg.Response, error)

+ type BrokerThrottleHook interface

+ OnThrottle func(meta BrokerMetadata, throttleInterval time.Duration, ...)

type Client

+ func (cl *Client) RequestSharded(ctx context.Context, req kmsg.Request) []ResponseShard

type ConsumerOpt

+ func FetchMinBytes(b int32) ConsumerOpt

+ type ErrLargeRespSize struct

+ Limit int32

+ Size int32

+ func (e *ErrLargeRespSize) Error() string

type Opt

+ func BrokerMaxReadBytes(v int32) Opt

+ func MinVersions(versions kversion.Versions) Opt

type Record

+ ProducerEpoch int16

+ ProducerID int64

+ type ResponseShard struct

+ Err error

+ Meta BrokerMetadata

+ Req kmsg.Request

+ Resp kmsg.Response

Oct 16, 2020

Changes in this version

+ var ErrAborting = errors.New("client is aborting buffered records")

+ var ErrAlreadyInTransaction = errors.New("invalid attempt to begin a transaction while already in a transaction")

+ var ErrBrokerDead = errors.New("broker has died - the broker id either migrated or no longer exists")

+ var ErrClientTooOld = errors.New("client is too old; this client does not know what to do with this")

+ var ErrCommitWithFatalID = errors.New("cannot commit with a fatal producer id; retry with an abort")

+ var ErrConnDead = errors.New("connection is dead")

+ var ErrCorrelationIDMismatch = errors.New("correlation ID mismatch")

+ var ErrInvalidResp = errors.New("invalid response")

+ var ErrInvalidRespSize = errors.New("invalid response size less than zero")

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

+ var ErrNoDial = errors.New("unable to dial the broker")

+ var ErrNoPartitionsAvailable = errors.New("no partitions available")

+ var ErrNoResp = errors.New("message was not replied to in a response")

+ var ErrNotGroup = errors.New("invalid group function call when not assigned a group")

+ var ErrNotInTransaction = errors.New("cannot produce record transactionally if not in a transaction")

+ var ErrNotTransactional = errors.New("invalid attempt to begin a transaction with a non-transactional client")

+ var ErrPartitionDeleted = errors.New("partition no longer exists")

+ var ErrRecordTimeout = errors.New("records have timed out before they were able to be produced")

+ var ErrUnknownBroker = errors.New("unknown broker")

+ var ErrUnknownRequestKey = errors.New("request key is unknown")

+ type Acks struct

+ func AllISRAcks() Acks

+ func LeaderAck() Acks

+ func NoAck() Acks

+ type Broker struct

+ func (b *Broker) Request(ctx context.Context, req kmsg.Request) (kmsg.Response, error)

+ type BrokerConnectHook interface

+ OnConnect func(meta BrokerMetadata, dialDur time.Duration, conn net.Conn, err error)

+ type BrokerDisconnectHook interface

+ OnDisconnect func(meta BrokerMetadata, conn net.Conn)

+ type BrokerMetadata struct

+ Host string

+ NodeID int32

+ Port int32

+ Rack *string

+ type BrokerReadHook interface

+ OnRead func(meta BrokerMetadata, key int16, bytesRead int, ...)

+ type BrokerWriteHook interface

+ OnWrite func(meta BrokerMetadata, key int16, bytesWritten int, ...)

+ type Client struct

+ func NewClient(opts ...Opt) (*Client, error)

+ func (cl *Client) AbortBufferedRecords(ctx context.Context) error

+ func (cl *Client) AssignGroup(group string, opts ...GroupOpt)

+ func (cl *Client) AssignGroupTransactSession(group string, opts ...GroupOpt) *GroupTransactSession

+ func (cl *Client) AssignPartitions(opts ...DirectConsumeOpt)

+ func (cl *Client) BeginTransaction() error

+ func (cl *Client) BlockingCommitOffsets(ctx context.Context, uncommitted map[string]map[int32]EpochOffset, ...)

+ func (cl *Client) Broker(id int) *Broker

+ func (cl *Client) Close()

+ func (cl *Client) CommitOffsets(ctx context.Context, uncommitted map[string]map[int32]EpochOffset, ...)

+ func (cl *Client) CommittedOffsets() map[string]map[int32]EpochOffset

+ func (cl *Client) DiscoveredBrokers() []*Broker

+ func (cl *Client) EndTransaction(ctx context.Context, commit TransactionEndTry) error

+ func (cl *Client) Flush(ctx context.Context) error

+ func (cl *Client) PollFetches(ctx context.Context) Fetches

+ func (cl *Client) Produce(ctx context.Context, r *Record, promise func(*Record, error)) error

+ func (cl *Client) Request(ctx context.Context, req kmsg.Request) (kmsg.Response, error)

+ func (cl *Client) SeedBrokers() []*Broker

+ func (cl *Client) SetOffsets(setOffsets map[string]map[int32]EpochOffset)

+ func (cl *Client) UncommittedOffsets() map[string]map[int32]EpochOffset

+ type CompressionCodec struct

+ func GzipCompression() CompressionCodec

+ func Lz4Compression() CompressionCodec

+ func NoCompression() CompressionCodec

+ func SnappyCompression() CompressionCodec

+ func ZstdCompression() CompressionCodec

+ func (c CompressionCodec) WithLevel(level int) CompressionCodec

+ type ConsumerOpt interface

+ func ConsumeResetOffset(offset Offset) ConsumerOpt

+ func FetchIsolationLevel(level IsolationLevel) ConsumerOpt

+ func FetchMaxBytes(b int32) ConsumerOpt

+ func FetchMaxPartitionBytes(b int32) ConsumerOpt

+ func FetchMaxWait(wait time.Duration) ConsumerOpt

+ func KeepControlRecords() ConsumerOpt

+ func Rack(rack string) ConsumerOpt

+ type DirectConsumeOpt interface

+ func ConsumePartitions(partitions map[string]map[int32]Offset) DirectConsumeOpt

+ func ConsumeTopics(offset Offset, topics ...string) DirectConsumeOpt

+ func ConsumeTopicsRegex() DirectConsumeOpt

+ type EpochOffset struct

+ Epoch int32

+ Offset int64

+ type ErrDataLoss struct

+ ConsumedTo int64

+ Partition int32

+ ResetTo int64

+ Topic string

+ func (e *ErrDataLoss) Error() string

+ type Fetch struct

+ Topics []FetchTopic

+ type FetchError struct

+ Err error

+ Partition int32

+ Topic string

+ type FetchPartition struct

+ Err error

+ HighWatermark int64

+ LastStableOffset int64

+ LogStartOffset int64

+ Partition int32

+ Records []*Record

+ type FetchTopic struct

+ Partitions []FetchPartition

+ Topic string

+ type Fetches []Fetch

+ func (fs Fetches) Errors() []FetchError

+ func (fs Fetches) RecordIter() *FetchesRecordIter

+ type FetchesRecordIter struct

+ func (i *FetchesRecordIter) Done() bool

+ func (i *FetchesRecordIter) Next() *Record

+ type GroupBalancer interface

+ func CooperativeStickyBalancer() GroupBalancer

+ func RangeBalancer() GroupBalancer

+ func RoundRobinBalancer() GroupBalancer

+ func StickyBalancer() GroupBalancer

+ type GroupOpt interface

+ func AutoCommitInterval(interval time.Duration) GroupOpt

+ func Balancers(balancers ...GroupBalancer) GroupOpt

+ func DisableAutoCommit() GroupOpt

+ func GroupTopics(topics ...string) GroupOpt

+ func GroupTopicsRegex() GroupOpt

+ func HeartbeatInterval(interval time.Duration) GroupOpt

+ func InstanceID(id string) GroupOpt

+ func OnAssigned(onAssigned func(context.Context, map[string][]int32)) GroupOpt

+ func OnLost(onLost func(context.Context, map[string][]int32)) GroupOpt

+ func OnRevoked(onRevoked func(context.Context, map[string][]int32)) GroupOpt

+ func RebalanceTimeout(timeout time.Duration) GroupOpt

+ func RequireStableFetchOffsets() GroupOpt

+ func SessionTimeout(timeout time.Duration) GroupOpt

+ type GroupTransactSession struct

+ func (s *GroupTransactSession) Begin() error

+ func (s *GroupTransactSession) End(ctx context.Context, commit TransactionEndTry) (bool, error)

+ type Hook interface

+ type IsolationLevel struct

+ func ReadCommitted() IsolationLevel

+ func ReadUncommitted() IsolationLevel

+ type LogLevel int8

+ const LogLevelDebug

+ const LogLevelError

+ const LogLevelInfo

+ const LogLevelNone

+ const LogLevelWarn

+ func (l LogLevel) String() string

+ type Logger interface

+ Level func() LogLevel

+ Log func(level LogLevel, msg string, keyvals ...interface{})

+ func BasicLogger(dst io.Writer, level LogLevel, prefixFn func() string) Logger

+ type Offset struct

+ func NewOffset() Offset

+ func (o Offset) At(at int64) Offset

+ func (o Offset) AtEnd() Offset

+ func (o Offset) AtStart() Offset

+ func (o Offset) Relative(n int64) Offset

+ func (o Offset) WithEpoch(e int32) Offset

+ type Opt interface

+ func AutoTopicCreation() Opt

+ func BrokerConnDeadRetries(n int) Opt

+ func BrokerMaxWriteBytes(v int32) Opt

+ func ClientID(id string) Opt

+ func ConnTimeoutOverhead(overhead time.Duration) Opt

+ func Dialer(fn func(ctx context.Context, network, host string) (net.Conn, error)) Opt

+ func DisableClientID() Opt

+ func MaxVersions(versions kversion.Versions) Opt

+ func MetadataMaxAge(age time.Duration) Opt

+ func MetadataMinAge(age time.Duration) Opt

+ func RequestRetries(n int) Opt

+ func RetryBackoff(backoff func(int) time.Duration) Opt

+ func RetryTimeout(t func(int16) time.Duration) Opt

+ func SASL(sasls ...sasl.Mechanism) Opt

+ func SeedBrokers(seeds ...string) Opt

+ func SoftwareNameAndVersion(name, version string) Opt

+ func WithHooks(hooks ...Hook) Opt

+ func WithLogger(l Logger) Opt

+ type Partitioner interface

+ ForTopic func(string) TopicPartitioner

+ func StickyKeyPartitioner(overrideHasher PartitionerHasher) Partitioner

+ func StickyPartitioner() Partitioner

+ type PartitionerHasher func([]byte, int) int

+ func KafkaHasher(hashFn func([]byte) uint32) PartitionerHasher

+ func SaramaHasher(hashFn func([]byte) uint32) PartitionerHasher

+ type ProducerOpt interface

+ func BatchCompression(preference ...CompressionCodec) ProducerOpt

+ func BatchMaxBytes(v int32) ProducerOpt

+ func Linger(linger time.Duration) ProducerOpt

+ func ManualFlushing() ProducerOpt

+ func MaxBufferedRecords(n int) ProducerOpt

+ func OnDataLoss(fn func(string, int32)) ProducerOpt

+ func ProduceRequestTimeout(limit time.Duration) ProducerOpt

+ func RecordPartitioner(partitioner Partitioner) ProducerOpt

+ func RecordTimeout(timeout time.Duration) ProducerOpt

+ func RequiredAcks(acks Acks) ProducerOpt

+ func StopOnDataLoss() ProducerOpt

+ func TransactionTimeout(timeout time.Duration) ProducerOpt

+ func TransactionalID(id string) ProducerOpt

+ type Record struct

+ Attrs RecordAttrs

+ Headers []RecordHeader

+ Key []byte

+ LeaderEpoch int32

+ Offset int64

+ Partition int32

+ Timestamp time.Time

+ Topic string

+ Value []byte

+ type RecordAttrs struct

+ func (a RecordAttrs) CompressionType() uint8

+ func (a RecordAttrs) IsControl() bool

+ func (a RecordAttrs) IsTransactional() bool

+ func (a RecordAttrs) TimestampType() int8

+ type RecordHeader struct

+ Key string

+ Value []byte

+ type TopicPartitioner interface

+ OnNewBatch func()

+ Partition func(r *Record, n int) int

+ RequiresConsistency func(*Record) bool

+ type TransactionEndTry bool

+ const TryAbort

+ const TryCommit