core

package

Go to main page

Versions in this module

v1

v1.0.6

Sep 6, 2026

v1.0.5

Sep 5, 2026

Changes in this version

+ func OpenState(ctx context.Context, path string) (*duckdb.DB, error)

+ type Marks struct

+ func NewMarks() *Marks

+ func (m *Marks) Advance(topic string, partition int32, mark Mark)

+ func (m *Marks) Each(fn func(topic string, partition int32, mark Mark))

+ func (m *Marks) Empty() bool

+ func (m *Marks) Get(topic string, partition int32) (Mark, bool)

+ func (m *Marks) Len() int

type Message

+ HighWatermark int64

type Metrics

+ ConsumerLag metric.Int64Gauge

+ StateCommitCount metric.Int64Counter

+ StateCommitLatency metric.Float64Histogram

+ StateSizeBytes metric.Int64Gauge

+ StateTableRows metric.Int64Gauge

+ type OffsetStat struct

+ LeaderEpoch int32

+ Offset int64

+ Partition int32

+ Topic string

+ type OffsetStore struct

+ func NewOffsetStore(conn adbc.Connection) *OffsetStore

+ func (s *OffsetStore) Init(ctx context.Context) error

+ func (s *OffsetStore) Load(ctx context.Context) (*Marks, error)

+ func (s *OffsetStore) Save(ctx context.Context, marks *Marks) error

+ type StateStats struct

+ Offsets []OffsetStat

+ Path string

+ SizeBytes int64

+ Tables []TableStat

+ func CollectStateStats(ctx context.Context, conn adbc.Connection, path string) (*StateStats, error)

+ type TableStat struct

+ Name string

+ Rows int64

type Turbine

+ func (t *Turbine) SyncState(ctx context.Context) error

type TurbineOption

+ func WithStateStats(fn func() (*StateStats, error)) TurbineOption

+ func WithStateStore(offsets offsetSaver, tx stateTx) TurbineOption

v1.0.4

Sep 2, 2026

Changes in this version

+ type Mark struct

+ LeaderEpoch int32

+ Offset int64

+ type MarkCommitter interface

+ CommitMarks func(marks map[string]map[int32]Mark) error

type Message

+ LeaderEpoch int32

v1.0.3

Sep 1, 2026

v1.0.2

Sep 1, 2026

v1.0.1

Sep 1, 2026

v1.0.0

Sep 1, 2026

Changes in this version

+ func InitCommands(conn adbc.Connection, c *config.Conf) error

+ func InitTables(conn adbc.Connection, c *config.Conf) error

+ func InitUDFs(c *config.Conf) error

+ type ErrorPolicy int

+ const PolicyDLQ

+ const PolicyIgnore

+ const PolicyRaise

+ func ParseErrorPolicy(s string) (ErrorPolicy, error)

+ type Handler interface

+ Init func(ctx context.Context) error

+ Invoke func(ctx context.Context) (arrow.Table, error)

+ Write func(msg []byte) error

+ type Message struct

+ Offset int64

+ Partition int32

+ Topic string

+ Value []byte

+ func (m Message) HasMetadata() bool

+ type MetadataWriter interface

+ WriteMessage func(msg Message) error

+ type Metrics struct

+ BatchProcessingLatency metric.Float64Histogram

+ ErrorCount metric.Int64Counter

+ MessageCount metric.Int64Counter

+ SinkFlushCount metric.Int64Counter

+ SinkFlushLatency metric.Float64Histogram

+ SinkFlushNumRows metric.Int64Gauge

+ SourceReadLatency metric.Float64Histogram

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

+ type PipelineErrorPolicies struct

+ DLQSink Sink

+ Policy ErrorPolicy

+ type Sink interface

+ Batch func() (arrow.Table, error)

+ Flush func() error

+ WriteTable func(batch arrow.Table) error

+ type Source interface

+ Close func() error

+ Commit func() error

+ Start func() error

+ Stream func() <-chan []Message

+ type Stats struct

+ NumErrors int

+ StartTime time.Time

+ 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 struct

+ func NewTurbine(source Source, handler Handler, sink Sink, batchSize int, ...) *Turbine

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

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

+ type TurbineOption func(turbine *Turbine)

+ func WithMetrics(m *Metrics) TurbineOption

+ func WithTurbineLogger(l *zap.Logger) TurbineOption

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL