kafka

Documentation

Index

• type KafkaSource
  • func NewKafkaSource(vertexInstance *dfv1.VertexInstance, writers map[string][]isb.BufferWriter, ...) (*KafkaSource, error)
  • func (r *KafkaSource) Ack(_ context.Context, offsets []isb.Offset) []error
  • func (r *KafkaSource) Close() error
  • func (r *KafkaSource) ForceStop()
  • func (r *KafkaSource) GetName() string
  • func (r *KafkaSource) GetPartitionIdx() int32
  • func (r *KafkaSource) NoAck(_ context.Context, _ []isb.Offset)
  • func (r *KafkaSource) Pending(_ context.Context) (int64, error)
  • func (r *KafkaSource) PublishSourceWatermarks(msgs []*isb.ReadMessage)
  • func (r *KafkaSource) Read(_ context.Context, count int64) ([]*isb.ReadMessage, error)
  • func (r *KafkaSource) Start() <-chan struct{}
  • func (r *KafkaSource) Stop()
• type Option
  • func WithBufferSize(s int) Option
  • func WithGroupName(gn string) Option
  • func WithLogger(l *zap.SugaredLogger) Option
  • func WithReadTimeOut(t time.Duration) Option

Types

type KafkaSource

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

func NewKafkaSource

func NewKafkaSource(
	vertexInstance *dfv1.VertexInstance,
	writers map[string][]isb.BufferWriter,
	fsd forward.ToWhichStepDecider,
	transformerApplier applier.SourceTransformApplier,
	fetchWM fetch.Fetcher,
	toVertexPublisherStores map[string]store.WatermarkStore,
	publishWMStores store.WatermarkStore,
	opts ...Option) (*KafkaSource, error)

NewKafkaSource returns a KafkaSource reader based on Kafka Consumer Group.

func (*KafkaSource) Ack

func (r *KafkaSource) Ack(_ context.Context, offsets []isb.Offset) []error

Ack acknowledges an array of offset.

func (*KafkaSource) Close

func (r *KafkaSource) Close() error

func (*KafkaSource) ForceStop

func (r *KafkaSource) ForceStop()

func (*KafkaSource) GetName

func (r *KafkaSource) GetName() string

func (*KafkaSource) GetPartitionIdx

func (r *KafkaSource) GetPartitionIdx() int32

GetPartitionIdx returns the partition number for the source vertex buffer Source is like a buffer with only one partition. So, we always return 0

func (*KafkaSource) NoAck

func (r *KafkaSource) NoAck(_ context.Context, _ []isb.Offset)

func (*KafkaSource) Pending

func (r *KafkaSource) Pending(_ context.Context) (int64, error)

func (*KafkaSource) PublishSourceWatermarks

func (r *KafkaSource) PublishSourceWatermarks(msgs []*isb.ReadMessage)

func (*KafkaSource) Read

func (r *KafkaSource) Read(_ context.Context, count int64) ([]*isb.ReadMessage, error)

func (*KafkaSource) Start

func (r *KafkaSource) Start() <-chan struct{}

func (*KafkaSource) Stop

func (r *KafkaSource) Stop()

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 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