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

v2

v2.0.7

Jan 5, 2024

Changes in this version

+ func CreateTopic(addr string, topic string, numPartitions, replicationFactor int) error

+ func DeleteTopic(addr string, topics ...string) error

+ func WithAllowPublishAutoTopicCreation(enable bool) broker.Option

+ func WithScramMechanism(algoName ScramAlgorithm, username, password string) broker.Option

+ func WithSubscribeAutoCreateTopic(topic string, numPartitions, replicationFactor int) broker.SubscribeOption

+ type BalancerName string

+ type ScramAlgorithm string

+ const ScramAlgorithmSHA256

+ const ScramAlgorithmSHA512

v2.0.6

Dec 1, 2023

v2.0.5

Nov 21, 2023

v2.0.4

Nov 16, 2023

Changes in this version

+ const Crc32Balancer

+ const HashBalancer

+ const LeastBytesBalancer

+ const Murmur2Balancer

+ const ReferenceHashBalancer

+ const RoundRobinBalancer

+ func NewBroker(opts ...broker.Option) broker.Broker

+ func WithAllowAutoTopicCreation(enable bool) broker.Option

+ func WithAsync(enable bool) broker.Option

+ func WithBatchBytes(by int64) broker.Option

+ func WithBatchSize(size int) broker.Option

+ func WithBatchTimeout(timeout time.Duration) broker.Option

+ func WithCommitInterval(interval time.Duration) broker.Option

+ func WithCrc32Balancer(consistent bool) broker.PublishOption

+ func WithDialer(cfg *kafkaGo.Dialer) broker.Option

+ func WithDialerTimeout(tm time.Duration) broker.Option

+ func WithEnableErrorLogger(enable bool) broker.Option

+ func WithEnableLogger(enable bool) broker.Option

+ func WithEnableOneTopicOneWriter(enable bool) broker.Option

+ func WithErrorLogger(l kafkaGo.Logger) broker.Option

+ func WithHashBalancer(hasher hash.Hash32) broker.PublishOption

+ func WithHeaders(headers map[string]interface{}) broker.PublishOption

+ func WithHeartbeatInterval(interval time.Duration) broker.Option

+ func WithLeastBytesBalancer() broker.PublishOption

+ func WithLogger(l kafkaGo.Logger) broker.Option

+ func WithMaxAttempts(cnt int) broker.Option

+ func WithMaxBytes(bytes int) broker.Option

+ func WithMaxWait(time time.Duration) broker.Option

+ func WithMessageKey(key []byte) broker.PublishOption

+ func WithMessageOffset(offset int64) broker.PublishOption

+ func WithMinBytes(bytes int) broker.Option

+ func WithMurmur2Balancer(consistent bool) broker.PublishOption

+ func WithPartitionWatchInterval(interval time.Duration) broker.Option

+ func WithPlainMechanism(username, password string) broker.Option

+ func WithPublishMaxAttempts(cnt int) broker.Option

+ func WithQueueCapacity(cap int) broker.Option

+ func WithReadLagInterval(interval time.Duration) broker.Option

+ func WithReadTimeout(timeout time.Duration) broker.Option

+ func WithReaderConfig(cfg kafkaGo.ReaderConfig) broker.Option

+ func WithRebalanceTimeout(timeout time.Duration) broker.Option

+ func WithReferenceHashBalancer(hasher hash.Hash32) broker.PublishOption

+ func WithRetentionTime(time time.Duration) broker.Option

+ func WithRetries(cnt int) broker.Option

+ func WithRoundRobinBalancer() broker.PublishOption

+ func WithSessionTimeout(timeout time.Duration) broker.Option

+ func WithStartOffset(offset int64) broker.Option

+ func WithWatchPartitionChanges(enable bool) broker.Option

+ func WithWriteTimeout(timeout time.Duration) broker.Option

+ func WithWriterConfig(cfg WriterConfig) broker.Option

+ type ErrorLogger struct

+ func (l ErrorLogger) Printf(msg string, args ...interface{})

+ type Logger struct

+ func (l Logger) Printf(msg string, args ...interface{})

+ type MessageCarrier struct

+ func NewMessageCarrier(msg *kafkaGo.Message) MessageCarrier

+ func (c MessageCarrier) Get(key string) string

+ func (c MessageCarrier) Keys() []string

+ func (c MessageCarrier) Set(key, val string)

+ type Writer struct

+ EnableOneTopicOneWriter bool

+ Writer *kafkaGo.Writer

+ Writers map[string]*kafkaGo.Writer

+ func NewWriter(enableOneTopicOneWriter bool) *Writer

+ func (w *Writer) Close()

+ func (w *Writer) CreateProducer(writerConfig WriterConfig, saslMechanism sasl.Mechanism, tlsConfig *tls.Config) *kafkaGo.Writer

+ type WriterConfig struct

+ AllowAutoTopicCreation bool

+ Async bool

+ Balancer kafkaGo.Balancer

+ BatchBytes int64

+ BatchSize int

+ BatchTimeout time.Duration

+ Brokers []string

+ ErrorLogger kafkaGo.Logger

+ Logger kafkaGo.Logger

+ MaxAttempts int

+ ReadTimeout time.Duration

+ RequiredAcks kafkaGo.RequiredAcks

+ WriteTimeout time.Duration