kafkacgo

Documentation

Overview

Package kafkacgo provides a simple abstraction to send and pull Apache Kafka messages. IMPORTANT: it depends on C implementation, CGO must be enabled to use this package.

Index

• type Consumer
  • func NewConsumer(urls, topics []string, groupID string, opts ...Option) (*Consumer, error)
  • func (c *Consumer) Close() error
  • func (c *Consumer) Receive() ([]byte, error)
• type Offset
• type Option
  • func WithAutoOffsetResetPolicy(p Offset) Option
  • func WithConfigParameter(key string, val kafka.ConfigValue) Option
  • func WithProduceChannelSize(size int) Option
  • func WithSessionTimeout(t time.Duration) Option
• type Producer
  • func NewProducer(urls []string, opts ...Option) (*Producer, error)
  • func (p *Producer) Close()
  • func (p *Producer) Send(topic string, msg []byte) error

Types

type Consumer

type Consumer struct {
	// contains filtered or unexported fields
}

Consumer represents a wrapper around kafka.Consumer.

func NewConsumer

func NewConsumer(urls, topics []string, groupID string, opts ...Option) (*Consumer, error)

NewConsumer creates a new instance of Consumer.

func (*Consumer) Close

func (c *Consumer) Close() error

Close cleans up Consumer's internal resources.

func (*Consumer) Receive

func (c *Consumer) Receive() ([]byte, error)

Receive reads one message from the Kafka; is blocked if no messages in the queue.

type Offset

type Offset string

Offset points to where Kafka should start to read messages from.

const (
	// OffsetLatest automatically reset the offset to the latest offset.
	OffsetLatest Offset = "latest"

	// OffsetEarliest automatically reset the offset to the earliest offset.
	OffsetEarliest Offset = "earliest"

	// OffsetNone throw an error to the consumerClient if no previous offset is found for the consumerClient's group.
	OffsetNone Offset = "none"
)

type Option

type Option func(*config)

Option is a type alias for a function that configures Kafka client.

func WithAutoOffsetResetPolicy

func WithAutoOffsetResetPolicy(p Offset) Option

WithAutoOffsetResetPolicy sets what to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server (e.g. because that data has been deleted).

func WithConfigParameter

func WithConfigParameter(key string, val kafka.ConfigValue) Option

WithConfigParameter extends kafka.ConfigMap with additional parameters. Parameters are listed at: * consumer: https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html * producer: https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html

func WithProduceChannelSize

func WithProduceChannelSize(size int) Option

WithProduceChannelSize sets the buffer size (in number of messages).

func WithSessionTimeout

func WithSessionTimeout(t time.Duration) Option

WithSessionTimeout sets the timeout used to detect client failures when using Kafka's group management facility. The client sends periodic heartbeats to indicate its liveness to the broker. If no heartbeats are received by the broker before the expiration of this session timeout, then the broker will remove this client from the group and initiate a rebalance. Note that the value must be in the allowable range as configured in the broker configuration by group.min.session.timeout.ms and group.max.session.timeout.ms.

type Producer

type Producer struct {
	// contains filtered or unexported fields
}

Producer represents a wrapper around kafka.Producer.

func NewProducer

func NewProducer(urls []string, opts ...Option) (*Producer, error)

NewProducer creates a new instance of Producer.

func (*Producer) Close

func (p *Producer) Close()

Close cleans up Producer's internal resources.

func (*Producer) Send

func (p *Producer) Send(topic string, msg []byte) error

Send sends a message to Kafka topic.