kafka

Documentation

Index

• Constants
• type Config
  • func (c *Config) Validate() error
• type SASL
• type Sink
  • func NewSink() *Sink
  • func (s *Sink) Category() api.Category
  • func (s *Sink) Config() interface{}
  • func (s *Sink) Consume(batch api.Batch) api.Result
  • func (s *Sink) Init(context api.Context) error
  • func (s *Sink) SetCodec(c codec.Codec)
  • func (s *Sink) Start() error
  • func (s *Sink) Stop()
  • func (s *Sink) String() string
  • func (s *Sink) Type() api.Type

Constants

const (
	BalanceHash       = "hash"
	BalanceRoundRobin = "roundRobin"
	BalanceLeastBytes = "leastBytes"

	CompressionGzip   = "gzip"
	CompressionSnappy = "snappy"
	CompressionLz4    = "lz4"
	CompressionZstd   = "zstd"

	SASLNoneType  = ""
	SASLPlainType = "plain"
	SASLSCRAMType = "scram"

	AlgorithmSHA256 = "sha256"
	AlgorithmSHA512 = "sha512"
)

const Type = "kafka"

Types

type Config

type Config struct {
	Brokers      []string      `yaml:"brokers,omitempty" validate:"required"`
	Topic        string        `yaml:"topic,omitempty" validate:"required" default:"loggie"`
	Balance      string        `yaml:"balance,omitempty" default:"roundRobin"`
	Compression  string        `yaml:"compression,omitempty" default:"gzip"`
	MaxAttempts  int           `yaml:"maxAttempts,omitempty"`
	BatchSize    int           `yaml:"batchSize,omitempty"`
	BatchBytes   int64         `yaml:"batchBytes,omitempty"`
	BatchTimeout time.Duration `yaml:"batchTimeout,omitempty"`
	ReadTimeout  time.Duration `yaml:"readTimeout,omitempty"`
	WriteTimeout time.Duration `yaml:"writeTimeout,omitempty"`
	RequiredAcks int           `yaml:"requiredAcks,omitempty"`
	SASL         SASL          `yaml:"sasl,omitempty"`
}

func (*Config) Validate

func (c *Config) Validate() error

type SASL

type SASL struct {
	Type      string `yaml:"type,omitempty"`
	UserName  string `yaml:"userName,omitempty"`
	Password  string `yaml:"password,omitempty"`
	Algorithm string `yaml:"algorithm,omitempty"`
}

type Sink

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

func NewSink

func NewSink() *Sink

func (*Sink) Category

func (s *Sink) Category() api.Category

func (*Sink) Config

func (s *Sink) Config() interface{}

func (*Sink) Consume

func (s *Sink) Consume(batch api.Batch) api.Result

func (*Sink) Init

func (s *Sink) Init(context api.Context) error

func (*Sink) SetCodec

func (s *Sink) SetCodec(c codec.Codec)

func (*Sink) Start

func (s *Sink) Start() error

func (*Sink) Stop

func (s *Sink) Stop()

func (*Sink) String

func (s *Sink) String() string

func (*Sink) Type

func (s *Sink) Type() api.Type