core

type ErrorPolicy ¶

type ErrorPolicy int

const (
	PolicyRaise ErrorPolicy = iota
	PolicyIgnore
	PolicyDLQ
)

func ParseErrorPolicy(s string) (ErrorPolicy, error)

type Handler ¶

type Handler interface {
	Init(ctx context.Context) error
	Write(msg []byte) error
	Invoke(ctx context.Context) (arrow.Table, error)
}

type Mark ¶ added in v1.0.4

type Mark struct {
	Offset      int64
	LeaderEpoch int32
}

Mark is the position of the last message the pipeline has finished with in one partition: written to the handler, or dropped by an error policy.

type MarkCommitter ¶ added in v1.0.4

type MarkCommitter interface {
	CommitMarks(marks map[string]map[int32]Mark) error
}

MarkCommitter is implemented by sources that can commit an explicit position. The pipeline prefers it to Commit, because a source that reads ahead of the pipeline -- the Kafka source polls into a buffer -- has fetched messages the pipeline has not processed, and committing "everything fetched" commits those too. Marks are keyed by topic then partition.

type Message ¶

type Message struct {
	Value     []byte
	Topic     string
	Partition int32
	Offset    int64
	// LeaderEpoch is the Kafka leader epoch the record was read under. It is
	// carried through so a commit can name it, which lets the broker detect
	// log truncation. Only meaningful when HasMetadata is true; a source with
	// no positions leaves it zero along with the rest.
	LeaderEpoch int32
}

Message is one record from a source, with whatever provenance the source knows about it. Only Kafka populates the metadata fields.

func (m Message) HasMetadata() bool

type MetadataWriter ¶

type MetadataWriter interface {
	WriteMessage(msg Message) error
}

MetadataWriter is implemented by handlers that can use a message's source metadata. Handlers that only need the payload implement Handler alone and the consume loop hands them the value.

type Metrics ¶

type Metrics struct {
	MessageCount           metric.Int64Counter
	ErrorCount             metric.Int64Counter
	SourceReadLatency      metric.Float64Histogram
	SinkFlushLatency       metric.Float64Histogram
	SinkFlushNumRows       metric.Int64Gauge
	SinkFlushCount         metric.Int64Counter
	BatchProcessingLatency metric.Float64Histogram
}

Metrics holds the instruments the pipeline records, mirroring the names, descriptions and units the Python engine exports.

func NewMetrics(mp metric.MeterProvider) (*Metrics, error)

type PipelineErrorPolicies ¶

type PipelineErrorPolicies struct {
	Policy  ErrorPolicy
	DLQSink Sink
}

type Sink ¶

type Sink interface {
	WriteTable(batch arrow.Table) error
	Flush() error
	Batch() (arrow.Table, error)
}

type Source ¶

type Source interface {
	Start() error
	Stream() <-chan []Message
	Commit() error
	Close() error
}

type Stats ¶

type Stats struct {
	StartTime time.Time
	NumErrors int
	// contains filtered or unexported fields
}

func (s *Stats) AddMessagesConsumed(n int64)

func (s *Stats) GetThroughput() float64

func (s *Stats) MessagesConsumed() int64

func (s *Stats) SetNumMessagesConsumed(num int64)

func (s *Stats) SetThroughput(throughput float64)

type Turbine ¶

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

func NewTurbine(
	source Source,
	handler Handler,
	sink Sink,
	batchSize int,
	flushInterval time.Duration,
	lock *sync.Mutex,
	policy PipelineErrorPolicies,
	opts ...TurbineOption,
) *Turbine

func (t *Turbine) ConsumeLoop(ctx context.Context, maxMsgs int) (stats *Stats, err error)

func (t *Turbine) StatusLoop(ctx context.Context) error

type TurbineOption ¶

type TurbineOption func(turbine *Turbine)

func WithMetrics(m *Metrics) TurbineOption

func WithTurbineLogger(l *zap.Logger) TurbineOption