Documentation
¶
Index ¶
- func Factory() (pipeline.AnyPlugin, pipeline.AnyConfig)
- type Config
- 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()
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
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
// >
// > 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:"oldest|newest"` // *
}
! config-params ^ config-params
type Plugin ¶
type Plugin struct {
// contains filtered or unexported fields
}
func (*Plugin) ConsumeClaim ¶
func (p *Plugin) ConsumeClaim(_ sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
Click to show internal directories.
Click to hide internal directories.