kafka

Documentation

Index

• type Option
  • func WithLogger(log *zap.SugaredLogger) Option
• type ToKafka
  • func NewToKafka(vertexInstance *dfv1.VertexInstance, fromBuffer isb.BufferReader, ...) (*ToKafka, error)
  • func (tk *ToKafka) Close() error
  • func (tk *ToKafka) ForceStop()
  • func (tk *ToKafka) GetName() string
  • func (tk *ToKafka) GetPartitionIdx() int32
  • func (tk *ToKafka) Start() <-chan struct{}
  • func (tk *ToKafka) Stop()
  • func (tk *ToKafka) Write(_ context.Context, messages []isb.Message) ([]isb.Offset, []error)

Types

type Option

type Option func(*ToKafka) error

func WithLogger

func WithLogger(log *zap.SugaredLogger) Option

type ToKafka

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

ToKafka produce the output to a kafka sinks.

func NewToKafka

func NewToKafka(vertexInstance *dfv1.VertexInstance,
	fromBuffer isb.BufferReader,
	fetchWatermark fetch.Fetcher,
	publishWatermark publish.Publisher,
	idleManager wmb.IdleManager,
	opts ...Option) (*ToKafka, error)

NewToKafka returns ToKafka type.

func (*ToKafka) Close

func (tk *ToKafka) Close() error

func (*ToKafka) ForceStop

func (tk *ToKafka) ForceStop()

ForceStop stops sinking

func (*ToKafka) GetName

func (tk *ToKafka) GetName() string

GetName returns the name.

func (*ToKafka) GetPartitionIdx

func (tk *ToKafka) GetPartitionIdx() int32

GetPartitionIdx returns the partition index. for sink it is always 0.

func (*ToKafka) Start

func (tk *ToKafka) Start() <-chan struct{}

Start starts sinking to kafka.

func (*ToKafka) Stop

func (tk *ToKafka) Stop()

Stop stops sinking

func (*ToKafka) Write

func (tk *ToKafka) Write(_ context.Context, messages []isb.Message) ([]isb.Offset, []error)

Write writes to the kafka topic.