partitionworker

type AllocationMode ¶

type AllocationMode string

const (
	AllocationModePool   AllocationMode = "pool"
	AllocationModeStatic AllocationMode = "static"
)

type Allocator ¶

type Allocator interface {
	AllocateWorker(string, int, *time.Duration) (*worker.Worker, interface{}, error)
	ReleaseWorker(interface{}, *worker.Worker) error
	Stop() error
}

type PooledWorkerAllocator ¶

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

PooledWorkerAllocator holds a shared pool of workers for all partitions to use. cannot guarantee that single worker will always be used to handle the same partition but offers the best throughput since all partitions of a given topic share the same pool of workers

func NewPooledWorkerAllocator(parentLogger logger.Logger,
	workerAllocator worker.Allocator) (*PooledWorkerAllocator, error)

func (wa *PooledWorkerAllocator) AllocateWorker(topic string,
	partitionID int,
	timeout *time.Duration) (*worker.Worker, interface{}, error)

func (wa *PooledWorkerAllocator) ReleaseWorker(cookie interface{}, workerInstance *worker.Worker) error

func (wa *PooledWorkerAllocator) Stop() error

type StaticWorkerAllocator ¶

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

StaticWorkerAllocator statically maps a given partition to a given worker. this guarantees that a given partition in a given topic will *always* be handled by the same worker of this replica. for functions that benefit from holding in-order state this will be useful. however, the cost is throughput - it segments the worker pool such that it's possible that a partition mapped to a busy worker will wait processing an event even though there are free workers (which are mapped to other partitions)

func NewStaticWorkerAllocator(parentLogger logger.Logger,
	workerAllocator worker.Allocator,
	topicPartitionIDs map[string][]int) (*StaticWorkerAllocator, error)

func (wa *StaticWorkerAllocator) AllocateWorker(topic string,
	partitionID int,
	timeout *time.Duration) (*worker.Worker, interface{}, error)

func (wa *StaticWorkerAllocator) ReleaseWorker(cookie interface{}, workerInstance *worker.Worker) error

func (wa *StaticWorkerAllocator) Stop() error