kafka

Documentation

Index

• Constants
• func Mechanism(saslType, userName, password, algo string) (sasl.Mechanism, error)
• type Config
  • func (c *Config) SetDefaults()
  • func (c *Config) Validate() error
• type RenderTopicFail
• type SASL
  • func (s *SASL) Validate() error
• 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"

Functions

func Mechanism

func Mechanism(saslType, userName, password, algo string) (sasl.Mechanism, error)

Types

type Config

type Config struct {
	Brokers                       []string        `yaml:"brokers,omitempty" validate:"required"`
	Topic                         string          `yaml:"topic,omitempty" validate:"required" default:"loggie"`
	IfRenderTopicFailed           RenderTopicFail `yaml:"ifRenderTopicFailed,omitempty"`
	IgnoreUnknownTopicOrPartition bool            `yaml:"ignoreUnknownTopicOrPartition,omitempty"`
	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"`
	PartitionKey                  string          `yaml:"partitionKey,omitempty"`
}

func (*Config) SetDefaults

func (c *Config) SetDefaults()

func (*Config) Validate

func (c *Config) Validate() error

type RenderTopicFail

type RenderTopicFail struct {
	DropEvent    bool   `yaml:"dropEvent,omitempty" default:"true"`
	IgnoreError  bool   `yaml:"ignoreError,omitempty"`
	DefaultTopic string `yaml:"defaultTopic,omitempty"`
}

type SASL

type SASL struct {
	Type      string `yaml:"type,omitempty"`
	UserName  string `yaml:"userName,omitempty"` // Deprecated, use username instead
	Username  string `yaml:"username,omitempty"`
	Password  string `yaml:"password,omitempty"`
	Algorithm string `yaml:"algorithm,omitempty"`
}

func (*SASL) Validate

func (s *SASL) Validate() error

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