Versions in this module Expand all Collapse all v2 v2.0.1 Aug 9, 2023 v2.0.0 Jun 23, 2023 Changes in this version + type Buffer interface + Add func(record *Record) error + Close func() error + Flush func() error + Init func() error + Records func() []*Record + Reset func(dueTo error) error + type BufferConfig struct + FlushInterval time.Duration + Size int + type FailedMessageHandler func(err error, record kafka.Record) + type Generator struct + func (a *Generator) Generate(tps []kafka.TopicPartition, topologyBuilder topology.Topology) TaskGeneration + type OnFlush func(records []*Record) error + type Record struct + func NewTaskRecord(record kafka.Record) *Record + type Task interface + ID func() TaskID + Init func() error + Ready func() error + Restore func() error + Start func(ctx context.Context, claim kafka.PartitionClaim, ...) + Stop func() error + Store func(name string) topology.StateStore + Sync func() error + type TaskAssignment []*TaskMapping + func (a TaskAssignment) String() string + type TaskContext struct + func (ctx TaskContext) TaskID() string + type TaskGeneration struct + func (g *TaskGeneration) Assign(assignment ...kafka.TopicPartition) TaskAssignment + func (g *TaskGeneration) FindMappingByTP(partition kafka.TopicPartition) *TaskMapping + func (g *TaskGeneration) Mappings() TaskAssignment + type TaskID interface + Partition func() int32 + String func() string + Topics func() string + UniqueID func() string + type TaskManager interface + AddGlobalTask func(ctx topology.BuilderContext, id TaskID, topology topology.SubTopologyBuilder) (Task, error) + AddTask func(ctx topology.BuilderContext, id TaskID, topology topology.SubTopologyBuilder, ...) (Task, error) + NewTaskId func(prefix string, tp kafka.TopicPartition) TaskID + RemoveTask func(id TaskID) error + StoreInstances func(name string) []topology.StateStore + Task func(id TaskID) (Task, error) + func NewTaskManager(builderCtx topology.BuilderContext, logger log.Logger, ...) (TaskManager, error) + type TaskMapping struct + TPs []kafka.TopicPartition + func (a TaskMapping) SubTopologyBuilder() topology.SubTopologyBuilder + func (a TaskMapping) TaskId() TaskID + type TaskOpt func(*taskOptions) + func WithBufferFlushInterval(interval time.Duration) TaskOpt + func WithBufferSize(size int) TaskOpt + func WithFailedMessageHandler(handler FailedMessageHandler) TaskOpt