wmb

Documentation

Overview

Package wmb represents the offset-timeline pair and its corresponding encoder and decoder.

Index

• type WMB
  • func DecodeToWMB(b []byte) (WMB, error)
  • func (w WMB) EncodeToBytes() ([]byte, error)
• type Watermark
  • func (w Watermark) After(t time.Time) bool
  • func (w Watermark) Before(t time.Time) bool
  • func (w Watermark) String() string
  • func (w Watermark) UnixMilli() int64

Types

type WMB

type WMB struct {
	// Idle is set to true if the given processor entity hasn't published anything
	// to the offset timeline bucket in a batch processing cycle.
	// Idle is used to signal an idle watermark.
	Idle bool
	// Offset is the monotonically increasing index/offset of the buffer (buffer is the physical representation
	// of the partition of the edge).
	Offset int64
	// Watermark is tightly coupled with the offset and will be monotonically increasing for a given ProcessorEntity
	// as the offset increases.
	// When it is idling (Idle==true), for a given offset, the watermark can monotonically increase without offset
	// increasing.
	Watermark int64
}

WMB is used in the KV offset timeline bucket as the value for the given processor entity key.

func DecodeToWMB

func DecodeToWMB(b []byte) (WMB, error)

DecodeToWMB decodes the given byte array into a WMB object.

func (WMB) EncodeToBytes

func (w WMB) EncodeToBytes() ([]byte, error)

EncodeToBytes encodes a WMB object into byte array.

type Watermark

type Watermark time.Time

Watermark is the monotonically increasing watermark. It is tightly coupled with ProcessorEntitier as the processor is responsible for monotonically increasing Watermark for that processor. NOTE: today we support only second progression of watermark, we need to support millisecond too.

func (Watermark) After

func (w Watermark) After(t time.Time) bool

func (Watermark) Before

func (w Watermark) Before(t time.Time) bool

func (Watermark) String

func (w Watermark) String() string

func (Watermark) UnixMilli

func (w Watermark) UnixMilli() int64