task

package

v0.8.0 Latest Latest Go to latest Published: Oct 20, 2025 License: Apache-2.0 Imports: 30 Imported by: 0

Details

View Source

const (
	MaxCountInBuf  = 1 << 27
	MaxParallelism = 10
)

This section is empty.

This section is empty.

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

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

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

TaskService holds the configuration for each task

func NewTaskService(cfg *config.Config, taskCfg *config.TaskConfig, c *Consumer) (service *Service)

NewTaskService creates an instance of new tasks with kafka, clickhouse and paser instances

func (service *Service) Init() (err error)

Init initializes the kafka and clickhouse task associated with this service

func (service *Service) Put(msg *model.InputMessage, traceId string, flushFn func(traceId, with string)) error

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

func NewSharder(service *Service) (sh *Sharder, err error)

func (sh *Sharder) Calc(row *model.Row) (int, error)

func (sh *Sharder) Flush(c context.Context, wg *sync.WaitGroup, rmap map[int32]*model.BatchRange, traceId string)

func (sh *Sharder) PutElement(msgRow *model.MsgRow)

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

func NewShardingPolicy(shardingKey string, shardingStripe uint64, dims []*model.ColumnWithType, shards int) (policy *ShardingPolicy, err error)

func (policy *ShardingPolicy) Calc(row *model.Row) (shard int, err error)

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

Sinker object maintains number of task for each partition

func NewSinker(rcm cm.RemoteConfManager, http string, cmd *util.CmdOptions) *Sinker

NewSinker get an instance of sinker with the task list

func (s *Sinker) Close()

Close shutdown task

func (s *Sinker) GetCurrentConfig() *config.Config

func (s *Sinker) Init() (err error)

func (s *Sinker) Run()

Run is the mainloop to get and apply config