processor

Documentation

Overview

Package processor is the smallest processor entity for which the watermark will strictly monotonically increase.

Index

• type ProcessorEntitier
  • func NewProcessorEntity(name string) ProcessorEntitier
• type ProcessorHeartbeat
  • func NewProcessorHeartbeat() *ProcessorHeartbeat
  • func (hb *ProcessorHeartbeat) Delete(key string)
  • func (hb *ProcessorHeartbeat) Get(key string) int64
  • func (hb *ProcessorHeartbeat) GetAll() map[string]int64
  • func (hb *ProcessorHeartbeat) Put(key string, value int64)
• type ProcessorManager
  • func NewProcessorManager(ctx context.Context, watermarkStoreWatcher store.WatermarkStoreWatcher, ...) *ProcessorManager
  • func (v *ProcessorManager) AddProcessor(processor string, p *ProcessorToFetch)
  • func (v *ProcessorManager) DeleteProcessor(processor string)
  • func (v *ProcessorManager) GetAllProcessors() map[string]*ProcessorToFetch
  • func (v *ProcessorManager) GetProcessor(processor string) *ProcessorToFetch
• type ProcessorManagerOption
  • func WithIsReduce(isReduce bool) ProcessorManagerOption
  • func WithIsSource(isSource bool) ProcessorManagerOption
  • func WithPodHeartbeatRate(rate int64) ProcessorManagerOption
  • func WithRefreshingProcessorsRate(rate int64) ProcessorManagerOption
  • func WithVertexReplica(replica int32) ProcessorManagerOption
• type ProcessorToFetch
  • func NewProcessorToFetch(ctx context.Context, processor ProcessorEntitier, capacity int, ...) *ProcessorToFetch
  • func (p *ProcessorToFetch) GetEntity() ProcessorEntitier
  • func (p *ProcessorToFetch) GetOffsetTimelines() []*timeline.OffsetTimeline
  • func (p *ProcessorToFetch) IsActive() bool
  • func (p *ProcessorToFetch) IsDeleted() bool
  • func (p *ProcessorToFetch) IsInactive() bool
  • func (p *ProcessorToFetch) String() string

Types

type ProcessorEntitier

type ProcessorEntitier interface {
	GetName() string
}

ProcessorEntitier defines what can be a processor. The Processor is the smallest unit where the watermark will monotonically increase.

func NewProcessorEntity

func NewProcessorEntity(name string) ProcessorEntitier

NewProcessorEntity returns a new `ProcessorEntitier`.

type ProcessorHeartbeat

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

ProcessorHeartbeat has details about each processor heartbeat. This information is populated by watching the Vn-1th vertex's processors. It stores only the latest heartbeat value.

func NewProcessorHeartbeat

func NewProcessorHeartbeat() *ProcessorHeartbeat

NewProcessorHeartbeat returns ProcessorHeartbeat.

func (*ProcessorHeartbeat) Delete

func (hb *ProcessorHeartbeat) Delete(key string)

Delete deletes a processor from the ProcessorHeartbeat table.

func (*ProcessorHeartbeat) Get

func (hb *ProcessorHeartbeat) Get(key string) int64

Get gets the heartbeat for a given processor.

func (*ProcessorHeartbeat) GetAll

func (hb *ProcessorHeartbeat) GetAll() map[string]int64

GetAll returns all the heartbeat entries in the heartbeat table.

func (*ProcessorHeartbeat) Put

func (hb *ProcessorHeartbeat) Put(key string, value int64)

Put inserts a heartbeat entry for a given processor key and value.

type ProcessorManager

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

ProcessorManager manages the point of view of Vn-1 from Vn vertex processors (or source processor). The code is running on Vn vertex. It has the mapping of all the processors which in turn has all the information about each processor timelines.

func NewProcessorManager

func NewProcessorManager(ctx context.Context, watermarkStoreWatcher store.WatermarkStoreWatcher, fromBufferPartitionCount int32, inputOpts ...ProcessorManagerOption) *ProcessorManager

NewProcessorManager returns a new ProcessorManager instance

func (*ProcessorManager) AddProcessor

func (v *ProcessorManager) AddProcessor(processor string, p *ProcessorToFetch)

AddProcessor adds a new processor. If the given processor already exists, the value will be updated.

func (*ProcessorManager) DeleteProcessor

func (v *ProcessorManager) DeleteProcessor(processor string)

DeleteProcessor deletes a processor.

func (*ProcessorManager) GetAllProcessors

func (v *ProcessorManager) GetAllProcessors() map[string]*ProcessorToFetch

GetAllProcessors returns all the processors.

func (*ProcessorManager) GetProcessor

func (v *ProcessorManager) GetProcessor(processor string) *ProcessorToFetch

GetProcessor gets a processor.

type ProcessorManagerOption

type ProcessorManagerOption func(*processorManagerOptions)

ProcessorManagerOption set options for FromVertex.

func WithIsReduce

func WithIsReduce(isReduce bool) ProcessorManagerOption

WithIsReduce sets the processor manager is for reduce.

func WithIsSource

func WithIsSource(isSource bool) ProcessorManagerOption

WithIsSource sets the vertex is source.

func WithPodHeartbeatRate

func WithPodHeartbeatRate(rate int64) ProcessorManagerOption

WithPodHeartbeatRate sets the heartbeat rate in seconds.

func WithRefreshingProcessorsRate

func WithRefreshingProcessorsRate(rate int64) ProcessorManagerOption

WithRefreshingProcessorsRate sets the processor refreshing rate in seconds.

func WithVertexReplica

func WithVertexReplica(replica int32) ProcessorManagerOption

WithVertexReplica sets the vertex replica.

type ProcessorToFetch

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

ProcessorToFetch is the smallest unit of entity (from which we fetch data) that does inorder processing or contains inorder data. It tracks OT for all the partitions of the from buffer.

func NewProcessorToFetch

func NewProcessorToFetch(ctx context.Context, processor ProcessorEntitier, capacity int, fromBufferPartitionCount int32) *ProcessorToFetch

NewProcessorToFetch creates ProcessorToFetch.

func (*ProcessorToFetch) GetEntity

func (p *ProcessorToFetch) GetEntity() ProcessorEntitier

GetEntity returns the processor entity.

func (*ProcessorToFetch) GetOffsetTimelines

func (p *ProcessorToFetch) GetOffsetTimelines() []*timeline.OffsetTimeline

GetOffsetTimelines returns the processor's OT.

func (*ProcessorToFetch) IsActive

func (p *ProcessorToFetch) IsActive() bool

IsActive returns whether a processor is active.

func (*ProcessorToFetch) IsDeleted

func (p *ProcessorToFetch) IsDeleted() bool

IsDeleted returns whether a processor has been deleted.

func (*ProcessorToFetch) IsInactive

func (p *ProcessorToFetch) IsInactive() bool

IsInactive returns whether a processor is inactive (no heartbeats or any sort).

func (*ProcessorToFetch) String

func (p *ProcessorToFetch) String() string