Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

kafka

package module

Go to main page

Versions in this module

v1

v1.2.84

Jan 17, 2024

v1.2.50

Jan 12, 2024

v1.2.11

Dec 19, 2023

Changes in this version

+ const AlternativeTimestamp

+ const DefaultTimestamp

+ const UnknownTimestamp

+ func New(options ...Option) services.Listenable

+ func PublishAsync(ctx context.Context, message ...ProducerMessage) (err error)

+ type ConsumeErrorHandler func(topic string, partition int32, cause error)

+ type ConsumeHandler func(ctx context.Context, value []byte, meta Meta) (err error)

type Consumer

+ func NewGroupConsumer(log logs.Logger, maxPollRecords int, partitionBuffer int, opts []kgo.Opt, ...) (v Consumer, err error)

+ Listen func(ctx context.Context) (err error)

+ Shutdown func(ctx context.Context)

+ type ConsumerOptions struct

+ Config configs.ConsumerConfig

+ Log logs.Logger

+ type GroupConsumer struct

+ func (consumer *GroupConsumer) Listen(ctx context.Context) (err error)

+ func (consumer *GroupConsumer) Shutdown(ctx context.Context)

+ type GroupPartitionConsumer struct

+ func (consumer *GroupPartitionConsumer) Consume(ctx context.Context)

+ type Header struct

+ Key string

+ Value []byte

+ type Headers []Header

+ func (headers Headers) ConvertToKafkaHeaders() (v []kgo.RecordHeader)

+ type Meta struct

+ Headers Headers

+ IsControl bool

+ IsTransactional bool

+ Key []byte

+ LeaderEpoch int32

+ Offset int64

+ Partition int32

+ ProducerEpoch int16

+ ProducerId int64

+ Time time.Time

+ TimestampType TimestampType

+ Topic string

+ type Option func(options *Options)

+ func WithConsumeErrorHandler(handler ConsumeErrorHandler) Option

+ func WithConsumeHandler(name string, handler ConsumeHandler) Option

+ type Options struct

type Producer

+ func NewProducer(log logs.Logger, num int, opts []kgo.Opt) (v *Producer, err error)

+ func (producer *Producer) Client() (v *kgo.Client)

+ func (producer *Producer) Shutdown(_ context.Context)

+ type ProducerMessage struct

+ Body []byte

+ Headers Headers

+ Key []byte

+ Topic string

+ func NewMessage(topic string, key []byte, body []byte) ProducerMessage

+ func (msg ProducerMessage) AddHeader(key string, value []byte) ProducerMessage

+ func (msg ProducerMessage) Validate() (err error)

+ type TimestampType int8

+ type TopicPartitionKey struct

+ Partition int32

+ Topic string

v0

v0.12.5

Jul 6, 2022

v0.12.4

Jul 5, 2022

v0.12.3

Jul 2, 2022

v0.12.2

Jul 2, 2022

v0.12.1

Jun 22, 2022

Changes in this version

+ func Publish(ctx context.Context, argument PublishArgument) (ok bool, err errors.CodeError)

+ func RegisterConsumerHandler(name string, builder ConsumerHandlerBuilder)

+ func Service() service.Service

+ type ClientTLSConfig struct

+ CA string

+ Cert string

+ Key string

+ type Config struct

+ Brokers []string

+ Consumers map[string]*ConsumerConfig

+ Options *OptionsConfig

+ Producers map[string]*ProducerConfig

+ type Consumer struct

+ func (consumer *Consumer) Close() (err error)

+ func (consumer *Consumer) Consume(ctx context.Context) (err error)

+ type ConsumerConfig struct

+ AutoCommit bool

+ GroupId string

+ Handler string

+ HandlerOptions json.RawMessage

+ type ConsumerHandler interface

+ Handle func(ctx context.Context, message ConsumerMessage)

+ type ConsumerHandlerBuilder func(options ConsumerHandlerOptions) (handler ConsumerHandler, err error)

+ type ConsumerHandlerOptions struct

+ Config configuares.Config

+ Log logs.Logger

+ type ConsumerMessage interface

+ Body func() (body []byte)

+ Commit func(ctx context.Context) (err error)

+ HighWaterMark func() (v int64)

+ Key func() (key []byte)

+ Offset func() (offset int64)

+ Partition func() (no int)

+ Raw func() (raw kafka.Message)

+ type DefaultConsumerMessage struct

+ func (msg *DefaultConsumerMessage) Body() (body []byte)

+ func (msg *DefaultConsumerMessage) Commit(ctx context.Context) (err error)

+ func (msg *DefaultConsumerMessage) HighWaterMark() (v int64)

+ func (msg *DefaultConsumerMessage) Key() (key []byte)

+ func (msg *DefaultConsumerMessage) Offset() (offset int64)

+ func (msg *DefaultConsumerMessage) Partition() (no int)

+ func (msg *DefaultConsumerMessage) Raw() (raw kafka.Message)

+ type ExplicitMessageCommitter struct

+ func (committer *ExplicitMessageCommitter) Commit(ctx context.Context, msg kafka.Message) (err error)

+ type Message struct

+ Argument json.RawMessage

+ Fn string

+ Service string

+ type MessageCommitter interface

+ Commit func(ctx context.Context, msg kafka.Message) (err error)

+ type MessageFetcher struct

+ func (fetcher *MessageFetcher) Close() (err error)

+ func (fetcher *MessageFetcher) ReadMessage(ctx context.Context) (msg kafka.Message, err error)

+ type MessageReader interface

+ Close func() (err error)

+ ReadMessage func(ctx context.Context) (msg kafka.Message, err error)

+ type NoopMessageCommitter struct

+ func (committer *NoopMessageCommitter) Commit(ctx context.Context, msg kafka.Message) (err error)

+ type OptionsConfig struct

+ Algo string

+ ClientId string

+ ClientTLS *ClientTLSConfig

+ DualStack bool

+ Password string

+ SASLType string

+ TimeoutSeconds int

+ Username string

+ type Printf struct

+ Core logs.Logger

+ func (p *Printf) Printf(layout string, v ...interface{})

+ type Producer struct

+ func (producer *Producer) Close() (err error)

+ func (producer *Producer) Publish(ctx context.Context, msg kafka.Message) (ok bool, err errors.CodeError)

+ type ProducerConfig struct

+ Async bool

+ Balancer string

+ BatchSize int

+ Compression string

+ RequiredAck string

+ type PublishArgument struct

+ Body json.RawMessage

+ Key string

+ Topic string

+ type PublishResult struct

+ Succeed bool