kafka

Documentation

Index

• func NewKafkaSourcePlugin(config Config, schema []schema.StreamSchema) sources.DataSource
• type Config
• type SourcePlugin
  • func (p *SourcePlugin) Connect(ctx context.Context) error
  • func (p *SourcePlugin) Events() chan sources.MessageEvent
  • func (p *SourcePlugin) GetConfig() []kgo.Opt
  • func (p *SourcePlugin) Start()
  • func (p *SourcePlugin) Stop()

Functions

func NewKafkaSourcePlugin

func NewKafkaSourcePlugin(config Config, schema []schema.StreamSchema) sources.DataSource

Types

type Config

type Config struct {
	Brokers       []string `json:"brokers" yaml:"brokers"`
	Sasl          bool     `json:"sasl" yaml:"sasl"`
	SaslPassword  string   `json:"sasl_password" yaml:"sasl_password"`
	SaslUser      string   `json:"sasl_user" yaml:"sasl_user"`
	SaslMechanism string   `json:"sasl_mechanism" yaml:"sasl_mechanism"`
	ConsumerGroup string   `json:"consumer_group" yaml:"consumer_group"`
}

type SourcePlugin

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

func (*SourcePlugin) Connect

func (p *SourcePlugin) Connect(ctx context.Context) error

func (*SourcePlugin) Events

func (p *SourcePlugin) Events() chan sources.MessageEvent

func (*SourcePlugin) GetConfig

func (p *SourcePlugin) GetConfig() []kgo.Opt

func (*SourcePlugin) Start

func (p *SourcePlugin) Start()

func (*SourcePlugin) Stop

func (p *SourcePlugin) Stop()