kafka

Documentation

Overview

Package kafka implements the messaging interfaces against a Kafka broker.

Index

• Variables
• type Client
  • func NewClient(brokerAddresses []string, opts ...Option) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) Commit(ctx context.Context) error
  • func (c *Client) Ping(ctx context.Context) error
  • func (c *Client) Poll(ctx context.Context) ([]*event.Event, error)
  • func (c *Client) Publish(ctx context.Context, ev *event.Event) error
• type Option
  • func WithConsumerGroup(group string, topics ...string) Option

Variables

var ErrNotConsumer = errors.New("client is not configured to consume")

ErrNotConsumer is returned by the consumer methods on a Client that was not built with WithConsumerGroup.

Types

type Client

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

Client implements messaging.Publisher, and additionally messaging.Consumer when constructed with WithConsumerGroup.

func NewClient

func NewClient(brokerAddresses []string, opts ...Option) (*Client, error)

NewClient connects to the brokers at brokerAddresses.

Producing is idempotent by default in franz-go: the broker discards a duplicate caused by a retried produce request. Note this only spans a single producer session — it is not protection against the application publishing the same event again after a restart, which is what event.Event.ID is for.

func (*Client) Close

func (c *Client) Close() error

func (*Client) Commit

func (c *Client) Commit(ctx context.Context) error

Commit acknowledges every event returned by Poll so far.

func (*Client) Ping

func (c *Client) Ping(ctx context.Context) error

func (*Client) Poll

func (c *Client) Poll(ctx context.Context) ([]*event.Event, error)

Poll blocks until at least one event is available, ctx is cancelled, or the Client is closed.

func (*Client) Publish

func (c *Client) Publish(ctx context.Context, ev *event.Event) error

Publish writes the event synchronously and returns once the brokers have acknowledged it.

The event key becomes the record key, which is what determines the partition and therefore the ordering guarantee: events sharing a key land on the same partition and are consumed in publish order.

type Option

type Option func(*options)

Option configures a Client.

func WithConsumerGroup

func WithConsumerGroup(group string, topics ...string) Option

WithConsumerGroup makes the Client a member of the named consumer group, subscribed to topics.

Partitions are shared among the members of a group, so scaling a group beyond the partition count leaves the extra members idle. Independent groups each receive the full stream, which is how one topic fans out to several services.

Auto-committing is disabled: offsets advance only when Commit is called, so an event is redelivered if the process dies before it is processed.