task

Documentation

Index

• Variables
• func NewProcessError(cause error, node string) error
• func NewProductionError(cause error, node string) error
• func NewSerdeError(cause error) error
• type Factory
  • func NewTopologyTaskFactory(t *topology.Topology, logger logger.Logger, opts ...FactoryOption) (Factory, error)
• type FactoryOption
  • func WithTelemetry(tel *streamsotel.Telemetry) FactoryOption
• type Manager
  • func NewManager(factory Factory, producer kafka.Producer, logger logger.Logger) Manager
• type ProcessError
  • func AsProcessError(err error) (*ProcessError, bool)
  • func (e *ProcessError) Error() string
  • func (e *ProcessError) Unwrap() error
• type ProductionError
  • func AsProductionError(err error) (*ProductionError, bool)
  • func (e *ProductionError) Error() string
  • func (e *ProductionError) Unwrap() error
• type SerdeError
  • func AsSerdeError(err error) (*SerdeError, bool)
  • func (e *SerdeError) Error() string
  • func (e *SerdeError) Unwrap() error
• type Task
• type TopologyTask
  • func (t *TopologyTask) Close() error
  • func (t *TopologyTask) IsClosed() bool
  • func (t *TopologyTask) Partition() kafka.TopicPartition
  • func (t *TopologyTask) Process(ctx context.Context, rec kafka.ConsumerRecord) error

Variables

var (
	ErrTaskClosed = errors.New("task is closed")
)

Functions

func NewProcessError

func NewProcessError(cause error, node string) error

func NewProductionError

func NewProductionError(cause error, node string) error

func NewSerdeError

func NewSerdeError(cause error) error

Types

type Factory

type Factory interface {
	CreateTask(partition kafka.TopicPartition, producer kafka.Producer) (Task, error)
}

func NewTopologyTaskFactory

func NewTopologyTaskFactory(t *topology.Topology, logger logger.Logger, opts ...FactoryOption) (Factory, error)

type FactoryOption

type FactoryOption func(*topologyTaskFactory)

FactoryOption configures a topologyTaskFactory

func WithTelemetry

func WithTelemetry(tel *streamsotel.Telemetry) FactoryOption

WithTelemetry sets the telemetry instance for tasks created by this factory

type Manager

type Manager interface {
	CreateTasks(partitions []kafka.TopicPartition) error
	CloseTasks(partitions []kafka.TopicPartition) error
	DeleteTasks(partitions []kafka.TopicPartition) error

	Tasks() map[kafka.TopicPartition]Task
	TaskFor(partition kafka.TopicPartition) (Task, bool)

	Close() error
}

Manager handles task lifecycle

func NewManager

func NewManager(factory Factory, producer kafka.Producer, logger logger.Logger) Manager

type ProcessError

type ProcessError struct {
	Cause error
	Node  string
}

func AsProcessError

func AsProcessError(err error) (*ProcessError, bool)

func (*ProcessError) Error

func (e *ProcessError) Error() string

func (*ProcessError) Unwrap

func (e *ProcessError) Unwrap() error

type ProductionError

type ProductionError struct {
	Cause error
	Node  string
}

ProductionError wraps errors that occur during sink production.

func AsProductionError

func AsProductionError(err error) (*ProductionError, bool)

func (*ProductionError) Error

func (e *ProductionError) Error() string

func (*ProductionError) Unwrap

func (e *ProductionError) Unwrap() error

type SerdeError

type SerdeError struct {
	Cause error
}

SerdeError wraps errors that occur during key/value serialization or deserialization.

func AsSerdeError

func AsSerdeError(err error) (*SerdeError, bool)

func (*SerdeError) Error

func (e *SerdeError) Error() string

func (*SerdeError) Unwrap

func (e *SerdeError) Unwrap() error

type Task

type Task interface {
	Partition() kafka.TopicPartition
	Process(ctx context.Context, record kafka.ConsumerRecord) error
	Close() error
	IsClosed() bool
}

Task processes records for a single partition

type TopologyTask

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

func (*TopologyTask) Close

func (t *TopologyTask) Close() error

func (*TopologyTask) IsClosed

func (t *TopologyTask) IsClosed() bool

func (*TopologyTask) Partition

func (t *TopologyTask) Partition() kafka.TopicPartition

func (*TopologyTask) Process

func (t *TopologyTask) Process(ctx context.Context, rec kafka.ConsumerRecord) error