fetch

Documentation

Index

• func NewEdgeFetcher(ctx context.Context, wmStore store.WatermarkStore, ...) *edgeFetcher
• type Fetcher
  • func NewEdgeFetcherSet(ctx context.Context, vertexInstance *dfv1.VertexInstance, ...) Fetcher
• type HeadFetcher
• type Option
  • func WithIsReduce(isReduce bool) Option
  • func WithIsSource(isSource bool) Option
  • func WithPodHeartbeatRate(rate int64) Option
  • func WithRefreshingProcessorsRate(rate int64) Option
  • func WithVertexReplica(replica int32) Option
• type ProcessorToFetch
  • func NewProcessorToFetch(ctx context.Context, processor entity.ProcessorEntitier, capacity int, ...) *ProcessorToFetch
  • func (p *ProcessorToFetch) GetEntity() entity.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
• type SourceFetcher
  • func NewSourceFetcher(ctx context.Context, store store.WatermarkStore, opts ...Option) SourceFetcher

Functions

func NewEdgeFetcher

func NewEdgeFetcher(ctx context.Context, wmStore store.WatermarkStore, fromBufferPartitionCount int, opts ...Option) *edgeFetcher

NewEdgeFetcher returns a new edge fetcher. This could have been private, except that UI uses it.

Types

type Fetcher

type Fetcher interface {
	// ComputeWatermark computes a valid watermark for the given offset on the given partition
	ComputeWatermark(offset isb.Offset, fromPartitionIdx int32) wmb.Watermark
	// ComputeHeadIdleWMB computes a valid head idle WMB for the given partition.
	ComputeHeadIdleWMB(fromPartitionIdx int32) wmb.WMB
}

Fetcher fetches watermark data from Vn-1 vertex and computes the watermark for Vn.

func NewEdgeFetcherSet

func NewEdgeFetcherSet(ctx context.Context, vertexInstance *dfv1.VertexInstance, wmStores map[string]store.WatermarkStore, opts ...Option) Fetcher

NewEdgeFetcherSet creates a new edgeFetcherSet object which implements the Fetcher interface.

type HeadFetcher

type HeadFetcher interface {
	// ComputeHeadWatermark computes a valid head watermark for the given partition
	ComputeHeadWatermark(fromPartitionIdx int32) wmb.Watermark
}

HeadFetcher computes the watermark for Vn.

type Option

type Option func(options *options)

Option set options for FromVertex.

func WithIsReduce

func WithIsReduce(isReduce bool) Option

WithIsReduce to indicate if the vertex is reduce.

func WithIsSource

func WithIsSource(isSource bool) Option

WithIsSource to indicate if the vertex is source.

func WithPodHeartbeatRate

func WithPodHeartbeatRate(rate int64) Option

WithPodHeartbeatRate sets the heartbeat rate in seconds.

func WithRefreshingProcessorsRate

func WithRefreshingProcessorsRate(rate int64) Option

WithRefreshingProcessorsRate to set the rate of refreshing processors in seconds.

func WithVertexReplica

func WithVertexReplica(replica int32) Option

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 entity.ProcessorEntitier, capacity int, fromBufferPartitionCount int32) *ProcessorToFetch

NewProcessorToFetch creates ProcessorToFetch.

func (*ProcessorToFetch) GetEntity

func (p *ProcessorToFetch) GetEntity() entity.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

type SourceFetcher

type SourceFetcher interface {
	// ComputeWatermark computes the watermark, it will return the minimum of all the watermarks of the processors.
	ComputeWatermark() wmb.Watermark
	HeadFetcher
}

SourceFetcher fetches watermark data for source vertex.

func NewSourceFetcher

func NewSourceFetcher(ctx context.Context, store store.WatermarkStore, opts ...Option) SourceFetcher

NewSourceFetcher returns a new source fetcher, pm has the details about the processors responsible for writing to the buckets of the source buffer.