kafkachannel

Documentation

Index

• func FilterKafkaChannelOwnerByReferenceOrLabel() func(obj interface{}) bool
• func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
• func Shutdown()
• type Reconciler
  • func (r *Reconciler) ClearKafkaAdminClient(ctx context.Context)
  • func (r *Reconciler) FinalizeKind(ctx context.Context, channel *kafkav1beta1.KafkaChannel) reconciler.Event
  • func (r *Reconciler) ReconcileKind(ctx context.Context, channel *kafkav1beta1.KafkaChannel) reconciler.Event
  • func (r *Reconciler) SetKafkaAdminClient(ctx context.Context)

Functions

func FilterKafkaChannelOwnerByReferenceOrLabel

func FilterKafkaChannelOwnerByReferenceOrLabel() func(obj interface{}) bool

FilterKafkaChannelOwnerByReferenceOrLabel - Custom Filter For Common K8S Components "Owned" By KafkaChannels

This function is similar to, and based on, the various knative.dev/pkg/controller/FilterXYZ functions. It is used to filter common Kubernetes objects (Service, Deployment, etc) owned by KafkaChannels using either a K8S OwnerReference (preferred), or Name/Namespace marker labels. This secondary support for such marker labels is necessary to work around the need for Cross-Namespace OwnerReferences which are not supported by K8S.

func NewController

func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl

NewController Creates A New KafkaChannel Controller

func Shutdown

func Shutdown()

Shutdown - Graceful Shutdown Hook

Types

type Reconciler

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

Reconciler Implements controller.Reconciler for KafkaChannel Resources

func (*Reconciler) ClearKafkaAdminClient

func (r *Reconciler) ClearKafkaAdminClient(ctx context.Context)

ClearKafkaAdminClient Clears (Closes) The Reconciler's Kafka AdminClient

func (*Reconciler) FinalizeKind

func (r *Reconciler) FinalizeKind(ctx context.Context, channel *kafkav1beta1.KafkaChannel) reconciler.Event

FinalizeKind Implements The Finalizer Interface & Is Responsible For Performing The Finalization (Topic Deletion)

func (*Reconciler) ReconcileKind

func (r *Reconciler) ReconcileKind(ctx context.Context, channel *kafkav1beta1.KafkaChannel) reconciler.Event

ReconcileKind Implements The Reconciler Interface & Is Responsible For Performing The Reconciliation (Creation)

func (*Reconciler) SetKafkaAdminClient

func (r *Reconciler) SetKafkaAdminClient(ctx context.Context)

SetKafkaAdminClient Clears / Re-Sets The Kafka AdminClient On The Reconciler

Ideally we would re-use the Kafka AdminClient but due to Issues with the Sarama ClusterAdmin we're forced to recreate a new connection every time. We were seeing "broken-pipe" failures (non-recoverable) with the ClusterAdmin after periods of inactivity.

https://github.com/Shopify/sarama/issues/1162
https://github.com/Shopify/sarama/issues/866

EventHub AdminClients could be reused, and this is somewhat inefficient for them, but they are very simple lightweight REST clients so recreating them isn't a big deal and it simplifies the code significantly to not have to support both use cases.