kafka

Documentation

Index

• Constants
• type KafkaOutput
  • func NewKafkaOutput() *KafkaOutput
  • func (o *KafkaOutput) Configure(config core.StringMap) (err error)
  • func (o *KafkaOutput) Process(m *core.Message)
  • func (o *KafkaOutput) Start() (err error)
  • func (o *KafkaOutput) Stop()
• type KafkaOutputConfig

Constants

const (
	DefaultBatchCount        = 10
	DefaultMaxBytes          = 10 * 1024 * 1024
	DefaultHeartbeatInterval = 30
	DefaultSendTimeout       = 30 * time.Second
)

Types

type KafkaOutput

type KafkaOutput struct {
	*core.BaseOutput
	// contains filtered or unexported fields
}

func NewKafkaOutput

func NewKafkaOutput() *KafkaOutput

func (*KafkaOutput) Configure

func (o *KafkaOutput) Configure(config core.StringMap) (err error)

func (*KafkaOutput) Process

func (o *KafkaOutput) Process(m *core.Message)

func (*KafkaOutput) Start

func (o *KafkaOutput) Start() (err error)

func (*KafkaOutput) Stop

func (o *KafkaOutput) Stop()

type KafkaOutputConfig

type KafkaOutputConfig struct {
	ID              string
	KeyVariable     string // kafka partition key variable name
	TopicName       string // kafka topic name can be specified by TopicName or passed as a variable by
	TopicVariable   string // TopicVariable
	ServerAddresses []string
	FlushBatch      int    // SaramaConfig.Producer.Flush.Messages and SaramaConfig.Producer.Flush.MaxMessages
	MaxMessageBytes int    // SaramaConfig.Producer.MaxMessageBytes
	BlockOnHugeMsg  bool   // Whether block producing if message size too large error happens
	RequiredAcks    string // SaramaConfig.Producer.RequiredAcks
}