forward

Documentation

Overview

Package forward is used for creating a data forwarder for the UDF vertex.

Package forward does the Read (fromBufferPartition) -> Process (map UDF) -> Forward (toBuffers) -> Ack (fromBufferPartition) loop.

Index

• func DefaultOptions() *options
• type InterStepDataForward
  • func NewInterStepDataForward(vertexInstance *dfv1.VertexInstance, fromStep isb.BufferReader, ...) (*InterStepDataForward, error)
  • func (isdf *InterStepDataForward) ForceStop()
  • func (isdf *InterStepDataForward) IsShuttingDown() (bool, error)
  • func (isdf *InterStepDataForward) Start() <-chan error
  • func (isdf *InterStepDataForward) Stop()
• type Option
  • func WithCallbackUploader(cp *callback.Uploader) Option
  • func WithLogger(l *zap.SugaredLogger) Option
  • func WithReadBatchSize(f int64) Option
  • func WithRetryInterval(f time.Duration) Option
  • func WithUDFConcurrency(f int) Option
  • func WithUDFMap(f applier.MapApplier) Option
  • func WithUDFStreamingMap(f applier.MapStreamApplier) Option
• type Shutdown
  • func (s *Shutdown) String() string

Functions

func DefaultOptions

func DefaultOptions() *options

Types

type InterStepDataForward

type InterStepDataForward struct {
	Shutdown
	// contains filtered or unexported fields
}

InterStepDataForward forwards the data from previous step to the current step via inter-step buffer.

func NewInterStepDataForward

func NewInterStepDataForward(vertexInstance *dfv1.VertexInstance, fromStep isb.BufferReader, toSteps map[string][]isb.BufferWriter, fsd forwarder.ToWhichStepDecider, fetchWatermark fetch.Fetcher, publishWatermark map[string]publish.Publisher, idleManager wmb.IdleManager, opts ...Option) (*InterStepDataForward, error)

NewInterStepDataForward creates an inter-step forwarder.

func (*InterStepDataForward) ForceStop

func (isdf *InterStepDataForward) ForceStop()

ForceStop sets up the force shutdown flag.

func (*InterStepDataForward) IsShuttingDown

func (isdf *InterStepDataForward) IsShuttingDown() (bool, error)

IsShuttingDown returns whether we can stop processing.

func (*InterStepDataForward) Start

func (isdf *InterStepDataForward) Start() <-chan error

Start starts reading the buffer and forwards to the next buffers. Call `Stop` to stop.

func (*InterStepDataForward) Stop

func (isdf *InterStepDataForward) Stop()

Stop stops the processing.

type Option

type Option func(*options) error

func WithCallbackUploader

func WithCallbackUploader(cp *callback.Uploader) Option

WithCallbackUploader sets the callback uploader for the vertex

func WithLogger

func WithLogger(l *zap.SugaredLogger) Option

WithLogger is used to return logger information

func WithReadBatchSize

func WithReadBatchSize(f int64) Option

WithReadBatchSize sets the read batch size

func WithRetryInterval

func WithRetryInterval(f time.Duration) Option

WithRetryInterval sets the retry interval

func WithUDFConcurrency

func WithUDFConcurrency(f int) Option

WithUDFConcurrency sets concurrency for map UDF processing

func WithUDFMap

func WithUDFMap(f applier.MapApplier) Option

WithUDFMap enables the unary map for UDF if provided with a non-nil applier

func WithUDFStreamingMap

func WithUDFStreamingMap(f applier.MapStreamApplier) Option

WithUDFStreamingMap sets streaming for map UDF processing if provided with a non-nil applier

type Shutdown

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

Shutdown tracks and enforces the shutdown activity.

func (*Shutdown) String

func (s *Shutdown) String() string