consumergroup

Documentation

Index

• Constants
• Variables
• func Enqueue(userFacingResource string, enqueue func(key types.NamespacedName)) func(obj interface{})
• func Filter(userFacingResource string) func(obj interface{}) bool
• func NewController(ctx context.Context) *controller.Impl
• func ResyncOnStatefulSetChange(ctx context.Context, handle func(interface{}))
• type ConsumersPerPlacement
• type Reconciler
  • func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.ConsumerGroup) reconciler.Event
• type SchedulerConfig

Constants

const (
	// KafkaSourceScheduler is the key for the scheduler map for any KafkaSource.
	// Keep this constant lowercase.
	KafkaSourceScheduler = "kafkasource"
	// KafkaTriggerScheduler is the key for the scheduler map for any Kafka Trigger.
	// Keep this constant lowercase.
	KafkaTriggerScheduler = "trigger"
	// KafkaChannelScheduler is the key for the scheduler map for any KafkaChannel.
	// Keep this constant lowercase.
	KafkaChannelScheduler = "kafkachannel"

	// ConfigKafkaSchedulerName is the name of the ConfigMap to configure the scheduler.
	ConfigKafkaSchedulerName = "config-kafka-scheduler"
	// ConfigKafkaDeSchedulerName is the name of the ConfigMap to configure the descheduler.
	ConfigKafkaDeSchedulerName = "config-kafka-descheduler"
)

Variables

var (
	ErrNoSubscriberURI     = errors.New("no subscriber URI resolved")
	ErrNoDeadLetterSinkURI = errors.New("no dead letter sink URI resolved")
)

Functions

func Enqueue

func Enqueue(userFacingResource string, enqueue func(key types.NamespacedName)) func(obj interface{})

Enqueue enqueues using the provided enqueue function the resource associated with a ConsumerGroup

func Filter

func Filter(userFacingResource string) func(obj interface{}) bool

Filter returns a filter function based on the user-facing resource that a controller is tracking. Usable by FilteringResourceEventHandler.

func NewController

func NewController(ctx context.Context) *controller.Impl

func ResyncOnStatefulSetChange

func ResyncOnStatefulSetChange(ctx context.Context, handle func(interface{}))

Types

type ConsumersPerPlacement

type ConsumersPerPlacement struct {
	Placement *eventingduckv1alpha1.Placement
	Consumers []*kafkainternals.Consumer
}

type Reconciler

type Reconciler struct {
	SchedulerFunc   schedulerFunc
	ConsumerLister  kafkainternalslisters.ConsumerLister
	InternalsClient internalv1alpha1.InternalV1alpha1Interface

	NameGenerator names.NameGenerator

	SystemNamespace string
}

func (Reconciler) ReconcileKind

func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.ConsumerGroup) reconciler.Event

type SchedulerConfig

type SchedulerConfig struct {
	StatefulSetName     string
	RefreshPeriod       time.Duration
	Capacity            int32
	SchedulerPolicyType scheduler.SchedulerPolicyType
	SchedulerPolicy     *scheduler.SchedulerPolicy
	DeSchedulerPolicy   *scheduler.SchedulerPolicy
}