kafka

Documentation

Index

• Variables
• type KafkaMetric
• type KafkaSink
  • func NewSink(log logger.Underlying, registry go_metrics.Registry) (*KafkaSink, error)
  • func (s *KafkaSink) Close()
  • func (s *KafkaSink) HttpInfo() map[string]float64
  • func (s *KafkaSink) Init(ctx context.Context, format formats.Format, compression kt.Compression, ...) error
  • func (s *KafkaSink) Send(ctx context.Context, payload *kt.Output)

Variables

var (
	KafkaTopic            = flag.String("kafka_topic", "", "kafka topic to produce on")
	KafkaBootstrapServers = flag.String("bootstrap.servers", "", "bootstrap.servers")
)

Types

type KafkaMetric

type KafkaMetric struct {
	DeliveryErr go_metrics.Meter
	DeliveryWin go_metrics.Meter
}

type KafkaSink

type KafkaSink struct {
	logger.ContextL
	Topic string
	// contains filtered or unexported fields
}

func NewSink

func NewSink(log logger.Underlying, registry go_metrics.Registry) (*KafkaSink, error)

func (*KafkaSink) Close

func (s *KafkaSink) Close()

func (*KafkaSink) HttpInfo

func (s *KafkaSink) HttpInfo() map[string]float64

func (*KafkaSink) Init

func (s *KafkaSink) Init(ctx context.Context, format formats.Format, compression kt.Compression, fmtr formats.Formatter) error

func (*KafkaSink) Send

func (s *KafkaSink) Send(ctx context.Context, payload *kt.Output)