kafka

Documentation

Index

• func NewKafkaSource(vertexInstance *dfv1.VertexInstance, writers map[string][]isb.BufferWriter, ...) (sourcer.Sourcer, error)
• type Option
  • func WithBufferSize(s int) Option
  • func WithGroupName(gn string) Option
  • func WithHeaders() Option
  • func WithLogger(l *zap.SugaredLogger) Option
  • func WithReadTimeOut(t time.Duration) Option

Functions

func NewKafkaSource

func NewKafkaSource(
	vertexInstance *dfv1.VertexInstance,
	writers map[string][]isb.BufferWriter,
	fsd forwarder.ToWhichStepDecider,
	transformerApplier applier.SourceTransformApplier,
	fetchWM fetch.SourceFetcher,
	toVertexPublisherStores map[string]store.WatermarkStore,
	publishWMStores store.WatermarkStore,
	idleManager wmb.IdleManager,
	opts ...Option) (sourcer.Sourcer, error)

NewKafkaSource returns a kafkaSource reader based on Kafka Consumer Group.

Types

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 WithHeaders

func WithHeaders() Option

WithHeaders is used to return whether to include kafka headers in the payload. This will change the data format.

func WithLogger

func WithLogger(l *zap.SugaredLogger) Option

WithLogger is used to return logger information

func WithReadTimeOut

func WithReadTimeOut(t time.Duration) Option

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