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

kefka

package module

Go to main page

Versions in this module

v0

v0.5.0

Dec 21, 2022

v0.4.0

Dec 17, 2022

v0.3.0

Dec 16, 2022

v0.2.0

Dec 15, 2022

v0.1.0

Dec 11, 2022

Changes in this version

+ const DefaultPollTimeout

+ const FirstOffset

+ const LastOffset

+ func FlushAll(p kafkaProducer)

+ func LagForTopicPartition(client MetadataClient, topic string, partition int) (int64, error)

+ func MessageCount(client MetadataClient, topic string, partition int) (int64, error)

+ func ReadTopicPartitions(ctx context.Context, opts ReaderOptions) error

+ func SyncProduce(ctx context.Context, p kafkaProducer, topic string, key, value []byte) error

+ func SyncProduceMessage(ctx context.Context, p kafkaProducer, msg *kafka.Message) error

+ type CancelFunc func()

+ func Consume(consumer *kafka.Consumer, handler MessageHandler, errCb ErrorCallback) CancelFunc

+ type Commit func()

+ type Consumer struct

+ func NewConsumer(opts ConsumerOptions) (*Consumer, error)

+ func (c *Consumer) Assignments() ([]kafka.TopicPartition, error)

+ func (c *Consumer) Close() error

+ func (c *Consumer) Consume()

+ func (c *Consumer) Lag() (map[string]int64, error)

+ func (c *Consumer) LagForTopicPartition(topic string, partition int) (int64, error)

+ type ConsumerOptions struct

+ ErrorHandler ErrorCallback

+ Handler MessageHandler

+ KafkaConfig *kafka.ConfigMap

+ Logger Logger

+ PollTimeout time.Duration

+ Topic string

+ type ErrorCallback func(err error)

+ type LogLevel uint

+ const DebugLevel

+ const ErrorLevel

+ const FatalLevel

+ const InfoLevel

+ const WarnLevel

+ type Logger interface

+ Printf func(lvl LogLevel, format string, args ...any)

+ type LoggerFunc func(lvl LogLevel, format string, args ...any)

+ func (l LoggerFunc) Printf(level LogLevel, msg string, args ...any)

+ type MarshallFunc func(v any) ([]byte, error)

+ func JsonMarshaller() MarshallFunc

+ func MsgpackMarshaller() MarshallFunc

+ func StringMarshaller() MarshallFunc

+ type MessageHandler interface

+ Handle func(msg *kafka.Message, ack Commit)

+ type MessageHandlerFunc func(msg *kafka.Message, ack Commit)

+ func (m MessageHandlerFunc) Handle(msg *kafka.Message, ack Commit)

+ type MetadataClient interface

+ Assignment func() (partitions []kafka.TopicPartition, err error)

+ Committed func(partitions []kafka.TopicPartition, timeoutMs int) (offsets []kafka.TopicPartition, err error)

+ QueryWatermarkOffsets func(topic string, partition int32, timeoutMs int) (low int64, high int64, err error)

+ type OffsetWatermarks struct

+ Error error

+ High int64

+ Low int64

+ type PartitionEOFCallback func(topic string, partition int, offset int64)

+ type Producer struct

+ func NewProducer(opts ProducerOptions) (*Producer, error)

+ func (p *Producer) Close()

+ func (p *Producer) Flush(timeout time.Duration) int

+ func (p *Producer) Produce(topic string, key any, val any, deliveryChan chan kafka.Event) error

+ func (p *Producer) ProduceMessage(msg *kafka.Message, deliveryChan chan kafka.Event) error

+ func (p *Producer) SyncProduce(ctx context.Context, topic string, key any, val any) error

+ func (p *Producer) SyncProduceMessage(ctx context.Context, msg *kafka.Message) error

+ type ProducerOptions struct

+ KafkaConfig *kafka.ConfigMap

+ KeyMarshaller MarshallFunc

+ Logger Logger

+ ValueMarshaller MarshallFunc

+ type Reader struct

+ func NewReader(opts ReaderOptions) (*Reader, error)

+ func (r *Reader) Close() error

+ func (r *Reader) QueryWatermarkOffsets() map[string]OffsetWatermarks

+ func (r *Reader) Read()

+ type ReaderOptions struct

+ ErrorCallback ErrorCallback

+ KafkaConfig *kafka.ConfigMap

+ Logger Logger

+ MessageHandler MessageHandler

+ PartitionEOFCallback PartitionEOFCallback

+ PollTimeout time.Duration

+ TopicPartitions kafka.TopicPartitions

+ type UnmarshallFunc func(data []byte, v any) error

+ func JsonUnmarshaller() UnmarshallFunc

+ func MsgpackUnmarshaller() UnmarshallFunc

Other modules containing this package

github.com/jkratz55/kefka/v2