kafka

Documentation

Index

• type KafkaSource
  • func NewKafkaSource(vertexInstance *dfv1.VertexInstance, writers []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) Pending(ctx context.Context) (int64, error)
  • 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 []isb.BufferWriter,
	fsd forward.ToWhichStepDecider,
	mapApplier applier.MapApplier,
	fetchWM fetch.Fetcher,
	publishWM map[string]publish.Publisher,
	publishWMStores store.WatermarkStorer,
	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) Pending

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

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