kafka

Documentation

Index

• type BatchWriter
  • func NewBatchWriter(config *Config, opts ...Option) (*BatchWriter, error)
  • func (w *BatchWriter) Close() error
  • func (w *BatchWriter) Name() string
  • func (w *BatchWriter) ProcessWALEvent(ctx context.Context, walEvent *wal.Event) (retErr error)
  • func (w *BatchWriter) Send(ctx context.Context) error
• type Config
• type Option
  • func WithCheckpoint(c checkpointer.Checkpoint) Option
  • func WithInstrumentation(i *otel.Instrumentation) Option
  • func WithLogger(l loglib.Logger) Option

Types

type BatchWriter

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

BatchWriter is a kafka writer that uses batches to send the data to the configured kafka topic.

func NewBatchWriter

func NewBatchWriter(config *Config, opts ...Option) (*BatchWriter, error)

func (*BatchWriter) Close

func (w *BatchWriter) Close() error

func (*BatchWriter) Name

func (w *BatchWriter) Name() string

func (*BatchWriter) ProcessWALEvent

func (w *BatchWriter) ProcessWALEvent(ctx context.Context, walEvent *wal.Event) (retErr error)

ProcessWalEvent is called on every new message from the wal

func (*BatchWriter) Send

func (w *BatchWriter) Send(ctx context.Context) error

type Config

type Config struct {
	Kafka kafka.ConnConfig
	// BatchTime is the max time interval at which the batch sending to kafka is
	// triggered. Defaults to 1s
	BatchTimeout time.Duration
	// BatchBytes is the max size in bytes for a given batch. When this size is
	// reached, the batch is sent to kafka. Defaults to 1572864 bytes.
	BatchBytes int64
	// BatchSize is the max number of messages to be sent per batch. When this
	// size is reached, the batch is sent to kafka. Defaults to 100.
	BatchSize int
	// MaxQueueBytes is the max memory used by the batch writer for inflight
	// batches. Defaults to 100MiB
	MaxQueueBytes int64
}

type Option

type Option func(*BatchWriter)

func WithCheckpoint

func WithCheckpoint(c checkpointer.Checkpoint) Option

func WithInstrumentation

func WithInstrumentation(i *otel.Instrumentation) Option

func WithLogger

func WithLogger(l loglib.Logger) Option