kafka

Documentation

Index

• func NewKafkaSource(ctx context.Context, vertexInstance *dfv1.VertexInstance, ...) (sourcer.SourceReader, error)
• type ConsumerHandler
  • func NewConsumerHandler(readChanSize int) *ConsumerHandler
  • func (consumer *ConsumerHandler) Cleanup(sess sarama.ConsumerGroupSession) error
  • func (consumer *ConsumerHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (consumer *ConsumerHandler) Setup(sess sarama.ConsumerGroupSession) error
• type Option
  • func WithBufferSize(s int) Option
  • func WithGroupName(gn string) Option
  • func WithReadTimeOut(t time.Duration) Option

Functions

func NewKafkaSource

func NewKafkaSource(ctx context.Context, vertexInstance *dfv1.VertexInstance, handler *ConsumerHandler, opts ...Option) (sourcer.SourceReader, error)

NewKafkaSource returns a kafkaSource reader based on Kafka Consumer Group.

Types

type ConsumerHandler

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

ConsumerHandler struct

func NewConsumerHandler

func NewConsumerHandler(readChanSize int) *ConsumerHandler

NewConsumerHandler creates new handler and initializes the channel for passing messages

func (*ConsumerHandler) Cleanup

func (consumer *ConsumerHandler) Cleanup(sess sarama.ConsumerGroupSession) error

Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited

func (*ConsumerHandler) ConsumeClaim

func (consumer *ConsumerHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().

func (*ConsumerHandler) Setup

func (consumer *ConsumerHandler) Setup(sess sarama.ConsumerGroupSession) error

Setup is run at the beginning of a new session, before ConsumeClaim

type Option

type Option func(*kafkaSource) error

func WithBufferSize

func WithBufferSize(s int) Option

WithBufferSize is used to return size of message channel information

func WithGroupName

func WithGroupName(gn string) Option

WithGroupName is used to set the group name

func WithReadTimeOut

func WithReadTimeOut(t time.Duration) Option

WithReadTimeOut is used to set the read timeout for the from buffer