kafka

package
v1.0.16 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Dec 2, 2024 License: GPL-3.0 Imports: 16 Imported by: 0

Documentation

Index

Constants

View Source
const SinkType = "kafka"

Variables

This section is empty.

Functions

func Init

func Init(config *Config) (*sarama.Config, error)

func NewSyncProducer

func NewSyncProducer(config *Config) (sarama.SyncProducer, error)

Types

type CompressionStrategy

type CompressionStrategy string
var (
	CompressionStrategyNone   CompressionStrategy = "none"
	CompressionStrategyGZIP   CompressionStrategy = "gzip"
	CompressionStrategySnappy CompressionStrategy = "snappy"
	CompressionStrategyLZ4    CompressionStrategy = "lz4"
	CompressionStrategyZSTD   CompressionStrategy = "zstd"
)

type Config

type Config struct {
	Brokers         string              `yaml:"brokers"`
	Topic           string              `yaml:"topic"`
	TLS             bool                `yaml:"tls" default:"false"`
	TLSClientConfig TLSClientConfig     `yaml:"tlsClientConfig"`
	MaxQueueSize    int                 `yaml:"maxQueueSize" default:"51200"`
	FlushFrequency  time.Duration       `yaml:"flushFrequency" default:"10s"`
	FlushMessages   int                 `yaml:"flushMessages" default:"500"`
	FlushBytes      int                 `yaml:"flushBytes" default:"1000000"`
	MaxRetries      int                 `yaml:"maxRetries" default:"3"`
	Compression     CompressionStrategy `yaml:"compression" default:"none"`
	RequiredAcks    RequiredAcks        `yaml:"requiredAcks" default:"leader"`
	Partitioning    PartitionStrategy   `yaml:"partitioning" default:"none"`
}

func (*Config) Validate

func (c *Config) Validate() error

type ItemExporter

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

func NewItemExporter

func NewItemExporter(name string, config *Config, log logrus.FieldLogger) (ItemExporter, error)

func (ItemExporter) ExportItems

func (e ItemExporter) ExportItems(ctx context.Context, items []*xatu.DecoratedEvent) error

func (ItemExporter) Shutdown

func (e ItemExporter) Shutdown(ctx context.Context) error

type Kafka

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

func New

func New(name string, config *Config, log logrus.FieldLogger, filterConfig *xatu.EventFilterConfig, shippingMethod processor.ShippingMethod) (*Kafka, error)

func (*Kafka) HandleNewDecoratedEvent

func (h *Kafka) HandleNewDecoratedEvent(ctx context.Context, event *xatu.DecoratedEvent) error

func (*Kafka) HandleNewDecoratedEvents

func (h *Kafka) HandleNewDecoratedEvents(ctx context.Context, events []*xatu.DecoratedEvent) error

func (*Kafka) Name

func (h *Kafka) Name() string

func (*Kafka) Start

func (h *Kafka) Start(ctx context.Context) error

func (*Kafka) Stop

func (h *Kafka) Stop(ctx context.Context) error

func (*Kafka) Type

func (h *Kafka) Type() string

type PartitionStrategy

type PartitionStrategy string
var (
	PartitionStrategyNone   PartitionStrategy = "none"
	PartitionStrategyRandom PartitionStrategy = "random"
)

type RequiredAcks

type RequiredAcks string
var (
	RequiredAcksLeader RequiredAcks = "leader"
	RequiredAcksAll    RequiredAcks = "all"
	RequiredAcksNone   RequiredAcks = "none"
)

type TLSClientConfig added in v1.0.12

type TLSClientConfig struct {
	CertificatePath string `yaml:"certificatePath"`
	KeyPath         string `yaml:"keyPath"`
	CACertificate   string `yaml:"caCertificate"`
}

func (*TLSClientConfig) Validate added in v1.0.12

func (c *TLSClientConfig) Validate() error

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL