kafka

Documentation

Index

• Constants
• func Init()
• func NewKafkaObserver(properties map[string]interface{}) (observer.Observer, error)
• func NewKafkaSink(properties map[string]interface{}) (sink.Sink, error)
• type KafkaObserver
  • func (o *KafkaObserver) Observe() error
  • func (o *KafkaObserver) Stop()
  • func (o *KafkaObserver) Wait()
• type KafkaResult
• type KafkaSink
  • func (s *KafkaSink) GetStats() *sink.Stats
  • func (s *KafkaSink) Init(name string, fields []common.Field) error
  • func (s *KafkaSink) Write(headers []string, rows [][]interface{}, index int) error

Constants

const KAFKA_OB_TYPE = "kafka"

const KAFKA_SASL_TYPE_NONE = "none"

const KAFKA_SASL_TYPE_PLAIN = "plain"

const KAFKA_SASL_TYPE_SCRAM = "scram"

const KAFKA_SINK_TYPE = "kafka"

Functions

func Init

func Init()

func NewKafkaObserver

func NewKafkaObserver(properties map[string]interface{}) (observer.Observer, error)

func NewKafkaSink

func NewKafkaSink(properties map[string]interface{}) (sink.Sink, error)

Types

type KafkaObserver

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

func (*KafkaObserver) Observe

func (o *KafkaObserver) Observe() error

func (*KafkaObserver) Stop

func (o *KafkaObserver) Stop()

func (*KafkaObserver) Wait

func (o *KafkaObserver) Wait()

type KafkaResult

type KafkaResult map[string]interface{}

type KafkaSink

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

func (*KafkaSink) GetStats

func (s *KafkaSink) GetStats() *sink.Stats

func (*KafkaSink) Init

func (s *KafkaSink) Init(name string, fields []common.Field) error

func (*KafkaSink) Write

func (s *KafkaSink) Write(headers []string, rows [][]interface{}, index int) error