kafka

Documentation

Index

• func Factory() (pipeline.AnyPlugin, pipeline.AnyConfig)
• func NewConsumerGroup(c *Config, l *zap.SugaredLogger) sarama.ConsumerGroup
• type Config
• type OffsetType
• type Plugin
  • func (p *Plugin) Cleanup(sarama.ConsumerGroupSession) error
  • func (p *Plugin) Commit(event *pipeline.Event)
  • func (p *Plugin) ConsumeClaim(_ sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (p *Plugin) PassEvent(_ *pipeline.Event) bool
  • func (p *Plugin) Setup(session sarama.ConsumerGroupSession) error
  • func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.InputPluginParams)
  • func (p *Plugin) Stop()

Functions

func Factory

func Factory() (pipeline.AnyPlugin, pipeline.AnyConfig)

func NewConsumerGroup

func NewConsumerGroup(c *Config, l *zap.SugaredLogger) sarama.ConsumerGroup

Types

type Config

type Config struct {
	// > @3@4@5@6
	// >
	// > The name of kafka brokers to read from.
	Brokers []string `json:"brokers" required:"true"` // *

	// > @3@4@5@6
	// >
	// > The list of kafka topics to read from.
	Topics []string `json:"topics" required:"true"` // *

	// > @3@4@5@6
	// >
	// > The name of consumer group to use.
	ConsumerGroup string `json:"consumer_group" default:"file-d"` // *

	// > @3@4@5@6
	// >
	// > Kafka client ID.
	ClientID string `json:"client_id" default:"file-d"` // *

	// > @3@4@5@6
	// >
	// > The number of unprocessed messages in the buffer that are loaded in the background from kafka.
	ChannelBufferSize int `json:"channel_buffer_size" default:"256"` // *

	// > @3@4@5@6
	// >
	// > The newest and oldest values is used when a consumer starts but there is no committed offset for the assigned partition.
	// > * *`newest`* - set offset to the newest message
	// > * *`oldest`* - set offset to the oldest message
	Offset  string `json:"offset" default:"newest" options:"newest|oldest"` // *
	Offset_ OffsetType

	// > @3@4@5@6
	// >
	// > The maximum amount of time the consumer expects a message takes to process for the user.
	ConsumerMaxProcessingTime  cfg.Duration `json:"consumer_max_processing_time" default:"200ms" parse:"duration"` // *
	ConsumerMaxProcessingTime_ time.Duration

	// > @3@4@5@6
	// >
	// > The maximum amount of time the broker will wait for Consumer.Fetch.Min bytes to become available before it returns fewer than that anyways.
	ConsumerMaxWaitTime  cfg.Duration `json:"consumer_max_wait_time" default:"250ms" parse:"duration"` // *
	ConsumerMaxWaitTime_ time.Duration

	// > @3@4@5@6
	// >
	// > If set, the plugin will use SASL authentications mechanism.
	SaslEnabled bool `json:"is_sasl_enabled" default:"false"` // *

	// > @3@4@5@6
	// >
	// > SASL mechanism to use.
	SaslMechanism string `json:"sasl_mechanism" default:"SCRAM-SHA-512" options:"PLAIN|SCRAM-SHA-256|SCRAM-SHA-512"` // *

	// > @3@4@5@6
	// >
	// > SASL username.
	SaslUsername string `json:"sasl_username" default:"user"` // *

	// > @3@4@5@6
	// >
	// > SASL password.
	SaslPassword string `json:"sasl_password" default:"password"` // *

	// > @3@4@5@6
	// >
	// > If set, the plugin will use SSL/TLS connections method.
	SslEnabled bool `json:"is_ssl_enabled" default:"false"` // *

	// > @3@4@5@6
	// >
	// > If set, the plugin will skip SSL/TLS verification.
	SslSkipVerify bool `json:"ssl_skip_verify" default:"false"` // *

	// > @3@4@5@6
	// >
	// > Path or content of a PEM-encoded CA file.
	SslPem string `json:"pem_file" default:"/file.d/certs"` // *
}

! config-params ^ config-params

type OffsetType

type OffsetType byte

const (
	OffsetTypeNewest OffsetType = iota
	OffsetTypeOldest
)

type Plugin

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

func (*Plugin) Cleanup

func (p *Plugin) Cleanup(sarama.ConsumerGroupSession) error

func (*Plugin) Commit

func (p *Plugin) Commit(event *pipeline.Event)

func (*Plugin) ConsumeClaim

func (p *Plugin) ConsumeClaim(_ sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (*Plugin) PassEvent

func (p *Plugin) PassEvent(_ *pipeline.Event) bool

PassEvent decides pass or discard event.

func (*Plugin) Setup

func (p *Plugin) Setup(session sarama.ConsumerGroupSession) error

func (*Plugin) Start

func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.InputPluginParams)

func (*Plugin) Stop

func (p *Plugin) Stop()