Repository: knative-sandbox/eventing-kafka
Pull request #924 (2021-10-07 10:12:36)

[0.24] KafkaChannel dispatcher offset checking improvements
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Fixing some issues created during https://github.com/knative-sandbox/eventing-kafka/pull/886
- Applying the fixes from https://github.com/aliok/eventing-kafka/pull/1 (that was on main)
- Change offset checking loop
- Requeue channel when there's a failure in the loop

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
- 🐛  Fixed an issue with the offset checking in the dispatcher so that the reconciliation is more proper
```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


pierDipi: @googlebot I consent.
knative-prow-robot: @aliok: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | fef5d4a64a18b67f2a32c80bc2afb2434dc8eb38 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/924/pull-knative-sandbox-eventing-kafka-go-coverage/1446055859888066560) | false | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-unit-tests | fef5d4a64a18b67f2a32c80bc2afb2434dc8eb38 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/924/pull-knative-sandbox-eventing-kafka-unit-tests/1446055859858706432) | true | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/924). [Your PR dashboard](https://gubernator.knative.dev/pr/aliok).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @aliok: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 24cf0c169b8897fb3989fda8c3623ba3ababcc5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/924/pull-knative-sandbox-eventing-kafka-unit-tests/1446068213359054848) | true | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | 24cf0c169b8897fb3989fda8c3623ba3ababcc5f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/924/pull-knative-sandbox-eventing-kafka-go-coverage/1446068213392609280) | false | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/924). [Your PR dashboard](https://gubernator.knative.dev/pr/aliok).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
aliok: /test pull-knative-sandbox-eventing-kafka-unit-tests
aliok: /test pull-knative-sandbox-eventing-kafka-unit-tests

```
=== RUN   TestEnableSaramaLogging
==================
WARNING: DATA RACE
```

I think this test is disabled in later branches
aliok: /test pull-knative-sandbox-eventing-kafka-upgrade-tests

It is stuck

Patch:
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index 19ecd000e..52992ff2b 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -24,6 +24,7 @@ import (
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
@@ -157,7 +158,7 @@ func main() {
 		MetricsRegistry: ekConfig.Sarama.Config.MetricRegistry,
 		SaramaConfig:    ekConfig.Sarama.Config,
 	}
-	dispatcher = dispatch.NewDispatcher(dispatcherConfig, controlProtocolServer)
+	dispatcher = dispatch.NewDispatcher(dispatcherConfig, controlProtocolServer, func(ref types.NamespacedName) {})
 
 	// Watch The Secret For Changes
 	err = distributedcommonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, environment.ResyncPeriod, secretObserver)
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher.go b/pkg/channel/consolidated/dispatcher/dispatcher.go
index e1da740f3..32036a147 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher.go
@@ -73,7 +73,7 @@ type KafkaDispatcher struct {
 	logger    *zap.SugaredLogger
 }
 
-func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispatcher, error) {
+func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs, enqueue func(ref types.NamespacedName)) (*KafkaDispatcher, error) {
 
 	producer, err := sarama.NewSyncProducer(args.Brokers, args.Config.Sarama.Config)
 	if err != nil {
@@ -82,7 +82,7 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispat
 
 	dispatcher := &KafkaDispatcher{
 		dispatcher:           eventingchannels.NewMessageDispatcher(logging.FromContext(ctx).Desugar()),
-		kafkaConsumerFactory: consumer.NewConsumerGroupFactory(args.Brokers, args.Config.Sarama.Config, &consumer.KafkaConsumerGroupOffsetsChecker{}),
+		kafkaConsumerFactory: consumer.NewConsumerGroupFactory(args.Brokers, args.Config.Sarama.Config, &consumer.KafkaConsumerGroupOffsetsChecker{}, enqueue),
 		channelSubscriptions: make(map[types.NamespacedName]*KafkaSubscription),
 		subsConsumerGroups:   make(map[types.UID]sarama.ConsumerGroup),
 		subscriptions:        make(map[types.UID]Subscription),
@@ -297,7 +297,7 @@ func (d *KafkaDispatcher) subscribe(ctx context.Context, channelRef types.Namesp
 	}
 	d.logger.Debugw("Starting consumer group", zap.Any("channelRef", channelRef),
 		zap.Any("subscription", sub.UID), zap.String("topic", topicName), zap.String("consumer group", groupID))
-	consumerGroup, err := d.kafkaConsumerFactory.StartConsumerGroup(ctx, groupID, []string{topicName}, handler)
+	consumerGroup, err := d.kafkaConsumerFactory.StartConsumerGroup(ctx, groupID, []string{topicName}, handler, channelRef)
 
 	if err != nil {
 		// we can not create a consumer - logging that, with reason
diff --git a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
index ccebe1dec..cbcd0fe9c 100644
--- a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
@@ -106,15 +106,7 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
 		Config:    kafkaConfig.EventingKafka,
 		TopicFunc: utils.TopicName,
 	}
-	kafkaDispatcher, err := dispatcher.NewDispatcher(ctx, args)
-	if err != nil {
-		logger.Fatalw("Unable to create kafka dispatcher", zap.Error(err))
-	}
-	logger.Info("Starting the Kafka dispatcher")
-	logger.Infow("Kafka broker configuration", zap.Strings("Brokers", kafkaConfig.Brokers))
-
 	r := &Reconciler{
-		kafkaDispatcher:      kafkaDispatcher,
 		kafkaClientSet:       kafkaclientsetinjection.Get(ctx),
 		kafkachannelLister:   kafkaChannelInformer.Lister(),
 		kafkachannelInformer: kafkaChannelInformer.Informer(),
@@ -123,6 +115,15 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
 		return controller.Options{SkipStatusUpdates: true}
 	})
 
+	kafkaDispatcher, err := dispatcher.NewDispatcher(ctx, args, r.impl.EnqueueKey)
+	if err != nil {
+		logger.Fatalw("Unable to create kafka dispatcher", zap.Error(err))
+	}
+	logger.Info("Starting the Kafka dispatcher")
+	logger.Infow("Kafka broker configuration", zap.Strings("Brokers", kafkaConfig.Brokers))
+
+	r.kafkaDispatcher = kafkaDispatcher
+
 	logger.Info("Setting up event handlers")
 
 	// Watch for kafka channels.
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
index 4fdcf7124..9c96f373f 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
@@ -25,6 +25,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
 	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -180,7 +181,11 @@ func (r Reconciler) reconcile(ctx context.Context, channel *kafkav1beta1.KafkaCh
 	}
 
 	// Update The ConsumerGroups To Align With Current KafkaChannel Subscribers
-	failedSubscriptions := r.dispatcher.UpdateSubscriptions(ctx, subscribers)
+	channelRef := types.NamespacedName{
+		Namespace: channel.GetNamespace(),
+		Name:      channel.GetName(),
+	}
+	failedSubscriptions := r.dispatcher.UpdateSubscriptions(ctx, channelRef, subscribers)
 
 	// Update The KafkaChannel Subscribable Status Based On ConsumerGroup Creation Status
 	channel.Status.SubscribableStatus = r.createSubscribableStatus(channel.Spec.Subscribers, failedSubscriptions)
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index 496b0c42e..884f0bbb0 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -66,7 +66,7 @@ func NewSubscriberWrapper(subscriberSpec eventingduck.SubscriberSpec, groupId st
 type Dispatcher interface {
 	SecretChanged(ctx context.Context, secret *corev1.Secret)
 	Shutdown()
-	UpdateSubscriptions(ctx context.Context, subscriberSpecs []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error
+	UpdateSubscriptions(ctx context.Context, channelRef types.NamespacedName, subscriberSpecs []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error
 }
 
 // DispatcherImpl Is A Struct With Configuration & ConsumerGroup State
@@ -84,9 +84,9 @@ type DispatcherImpl struct {
 var _ Dispatcher = &DispatcherImpl{}
 
 // NewDispatcher Is The Dispatcher Constructor
-func NewDispatcher(dispatcherConfig DispatcherConfig, controlServer controlprotocol.ServerHandler) Dispatcher {
+func NewDispatcher(dispatcherConfig DispatcherConfig, controlServer controlprotocol.ServerHandler, enqueue func(ref types.NamespacedName)) Dispatcher {
 
-	consumerGroupManager := commonconsumer.NewConsumerGroupManager(dispatcherConfig.Logger, controlServer, dispatcherConfig.Brokers, dispatcherConfig.SaramaConfig, &commonconsumer.NoopConsumerGroupOffsetsChecker{})
+	consumerGroupManager := commonconsumer.NewConsumerGroupManager(dispatcherConfig.Logger, controlServer, dispatcherConfig.Brokers, dispatcherConfig.SaramaConfig, &commonconsumer.NoopConsumerGroupOffsetsChecker{}, enqueue)
 
 	// Create The DispatcherImpl With Specified Configuration
 	dispatcher := &DispatcherImpl{
@@ -121,7 +121,7 @@ func (d *DispatcherImpl) Shutdown() {
 }
 
 // UpdateSubscriptions manages the Dispatcher's Subscriptions to align with new state
-func (d *DispatcherImpl) UpdateSubscriptions(ctx context.Context, subscriberSpecs []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error {
+func (d *DispatcherImpl) UpdateSubscriptions(ctx context.Context, channelRef types.NamespacedName, subscriberSpecs []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error {
 
 	if d.SaramaConfig == nil {
 		d.Logger.Error("Dispatcher has no config!")
@@ -150,7 +150,7 @@ func (d *DispatcherImpl) UpdateSubscriptions(ctx context.Context, subscriberSpec
 
 			// Create/Start A New ConsumerGroup With Custom Handler
 			handler := NewHandler(logger, groupId, &subscriberSpec)
-			err := d.consumerMgr.StartConsumerGroup(ctx, groupId, []string{d.Topic}, handler)
+			err := d.consumerMgr.StartConsumerGroup(ctx, groupId, []string{d.Topic}, handler, channelRef)
 			if err != nil {
 
 				// Log & Return Failure
diff --git a/pkg/common/consumer/consumer_factory.go b/pkg/common/consumer/consumer_factory.go
index dabaebf43..4ef170cdf 100644
--- a/pkg/common/consumer/consumer_factory.go
+++ b/pkg/common/consumer/consumer_factory.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/types"
 	"knative.dev/pkg/logging"
 )
 
@@ -33,13 +34,14 @@ type consumeFunc func(ctx context.Context, topics []string, handler sarama.Consu
 
 // KafkaConsumerGroupFactory creates the ConsumerGroup and start consuming the specified topic
 type KafkaConsumerGroupFactory interface {
-	StartConsumerGroup(ctx context.Context, groupID string, topics []string, handler KafkaConsumerHandler, options ...SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error)
+	StartConsumerGroup(ctx context.Context, groupID string, topics []string, handler KafkaConsumerHandler, channelRef types.NamespacedName, options ...SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error)
 }
 
 type kafkaConsumerGroupFactoryImpl struct {
 	config         *sarama.Config
 	addrs          []string
 	offsetsChecker ConsumerGroupOffsetsChecker
+	enqueue        func(ref types.NamespacedName)
 }
 
 type customConsumerGroup struct {
@@ -66,7 +68,7 @@ func (c *customConsumerGroup) Close() error {
 var _ sarama.ConsumerGroup = (*customConsumerGroup)(nil)
 
 // StartConsumerGroup creates a new customConsumerGroup and starts a Consume goroutine on it
-func (c kafkaConsumerGroupFactoryImpl) StartConsumerGroup(ctx context.Context, groupID string, topics []string, handler KafkaConsumerHandler, options ...SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error) {
+func (c kafkaConsumerGroupFactoryImpl) StartConsumerGroup(ctx context.Context, groupID string, topics []string, handler KafkaConsumerHandler, channelRef types.NamespacedName, options ...SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error) {
 	logger := logging.FromContext(ctx)
 
 	consumerGroup, err := c.createConsumerGroup(groupID)
@@ -76,7 +78,7 @@ func (c kafkaConsumerGroupFactoryImpl) StartConsumerGroup(ctx context.Context, g
 	}
 
 	// Start the consumerGroup.Consume function in a separate goroutine
-	return c.startExistingConsumerGroup(groupID, consumerGroup, consumerGroup.Consume, topics, logger, handler, options...), nil
+	return c.startExistingConsumerGroup(groupID, consumerGroup, consumerGroup.Consume, topics, logger, handler, channelRef, options...), nil
 }
 
 // createConsumerGroup creates a Sarama ConsumerGroup using the newConsumerGroup wrapper, with the
@@ -94,6 +96,7 @@ func (c kafkaConsumerGroupFactoryImpl) startExistingConsumerGroup(
 	topics []string,
 	logger *zap.SugaredLogger,
 	handler KafkaConsumerHandler,
+	channelRef types.NamespacedName,
 	options ...SaramaConsumerHandlerOption) *customConsumerGroup {
 
 	errorCh := make(chan error, 10)
@@ -105,8 +108,10 @@ func (c kafkaConsumerGroupFactoryImpl) startExistingConsumerGroup(
 		// do not proceed until the check is done
 		err := c.offsetsChecker.WaitForOffsetsInitialization(ctx, groupID, topics, logger, c.addrs, c.config)
 		if err != nil {
-			logger.Errorw("error while checking if offsets are initialized", zap.Any("topics", topics), zap.String("groupId", groupID), zap.Error(err))
+			logger.Errorw("error while checking if offsets are initialized", zap.Any("topics", topics), zap.String("groupId", groupID), zap.String("channel", channelRef.String()), zap.Error(err))
 			errorCh <- err
+			c.enqueue(channelRef)
+			return
 		}
 
 		logger.Debugw("all offsets are initialized", zap.Any("topics", topics), zap.Any("groupID", groupID))
@@ -136,8 +141,8 @@ func (c kafkaConsumerGroupFactoryImpl) startExistingConsumerGroup(
 	return &customConsumerGroup{cancel, errorCh, saramaGroup, releasedCh}
 }
 
-func NewConsumerGroupFactory(addrs []string, config *sarama.Config, offsetsChecker ConsumerGroupOffsetsChecker) KafkaConsumerGroupFactory {
-	return kafkaConsumerGroupFactoryImpl{addrs: addrs, config: config, offsetsChecker: offsetsChecker}
+func NewConsumerGroupFactory(addrs []string, config *sarama.Config, offsetsChecker ConsumerGroupOffsetsChecker, enqueue func(ref types.NamespacedName)) KafkaConsumerGroupFactory {
+	return kafkaConsumerGroupFactoryImpl{addrs: addrs, config: config, offsetsChecker: offsetsChecker, enqueue: enqueue}
 }
 
 var _ KafkaConsumerGroupFactory = (*kafkaConsumerGroupFactoryImpl)(nil)
diff --git a/pkg/common/consumer/consumer_handler.go b/pkg/common/consumer/consumer_handler.go
index 8face040a..9fd22f189 100644
--- a/pkg/common/consumer/consumer_handler.go
+++ b/pkg/common/consumer/consumer_handler.go
@@ -24,6 +24,7 @@ import (
 	"github.com/Shopify/sarama"
 
 	"go.uber.org/zap"
+
 	kafkasarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
 )
 
@@ -99,7 +100,7 @@ func NewConsumerHandler(logger *zap.SugaredLogger, handler KafkaConsumerHandler,
 
 // Setup is run at the beginning of a new session, before ConsumeClaim
 func (consumer *SaramaConsumerHandler) Setup(session sarama.ConsumerGroupSession) error {
-	consumer.logger.Info("setting up handler")
+	consumer.logger.Info("setting up handler", zap.Any("claims", session.Claims()))
 	consumer.lifecycleListener.Setup(session)
 	return nil
 }
diff --git a/pkg/common/consumer/consumer_manager.go b/pkg/common/consumer/consumer_manager.go
index 1dd6ad8e9..fd548b107 100644
--- a/pkg/common/consumer/consumer_manager.go
+++ b/pkg/common/consumer/consumer_manager.go
@@ -41,6 +41,7 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/types"
 	ctrlservice "knative.dev/control-protocol/pkg/service"
 	"knative.dev/pkg/logging"
 
@@ -58,7 +59,7 @@ var GroupLockedError = fmt.Errorf("managed group lock failed: locked by a differ
 // KafkaConsumerGroupManager keeps track of Sarama consumer groups and handles messages from control-protocol clients
 type KafkaConsumerGroupManager interface {
 	Reconfigure(brokers []string, config *sarama.Config) error
-	StartConsumerGroup(ctx context.Context, groupId string, topics []string, handler KafkaConsumerHandler, options ...SaramaConsumerHandlerOption) error
+	StartConsumerGroup(ctx context.Context, groupId string, topics []string, handler KafkaConsumerHandler, ref types.NamespacedName, options ...SaramaConsumerHandlerOption) error
 	CloseConsumerGroup(groupId string) error
 	Errors(groupId string) <-chan error
 	IsManaged(groupId string) bool
@@ -84,13 +85,13 @@ type kafkaConsumerGroupManagerImpl struct {
 var _ KafkaConsumerGroupManager = (*kafkaConsumerGroupManagerImpl)(nil)
 
 // NewConsumerGroupManager returns a new kafkaConsumerGroupManagerImpl as a KafkaConsumerGroupManager interface
-func NewConsumerGroupManager(logger *zap.Logger, serverHandler controlprotocol.ServerHandler, brokers []string, config *sarama.Config, offsetsChecker ConsumerGroupOffsetsChecker) KafkaConsumerGroupManager {
+func NewConsumerGroupManager(logger *zap.Logger, serverHandler controlprotocol.ServerHandler, brokers []string, config *sarama.Config, offsetsChecker ConsumerGroupOffsetsChecker, enqueue func(ref types.NamespacedName)) KafkaConsumerGroupManager {
 
 	manager := &kafkaConsumerGroupManagerImpl{
 		logger:         logger,
 		server:         serverHandler,
 		groups:         make(groupMap),
-		factory:        &kafkaConsumerGroupFactoryImpl{addrs: brokers, config: config, offsetsChecker: offsetsChecker},
+		factory:        &kafkaConsumerGroupFactoryImpl{addrs: brokers, config: config, offsetsChecker: offsetsChecker, enqueue: enqueue},
 		groupLock:      sync.RWMutex{},
 		offsetsChecker: offsetsChecker,
 	}
@@ -164,7 +165,7 @@ func getInternalLockCommand(lock bool) *commands.CommandLock {
 
 // StartConsumerGroup uses the consumer factory to create a new ConsumerGroup, add it to the list
 // of managed groups (for start/stop functionality) and start the Consume loop.
-func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(ctx context.Context, groupId string, topics []string, handler KafkaConsumerHandler, options ...SaramaConsumerHandlerOption) error {
+func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(ctx context.Context, groupId string, topics []string, handler KafkaConsumerHandler, ref types.NamespacedName, options ...SaramaConsumerHandlerOption) error {
 	logger := logging.FromContext(ctx)
 
 	groupLogger := m.logger.With(zap.String("GroupId", groupId))
@@ -190,7 +191,7 @@ func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(ctx context.Context,
 	}
 
 	// The only thing we really want from the factory is the cancel function for the customConsumerGroup
-	customGroup := m.factory.startExistingConsumerGroup(groupId, group, consume, topics, logger, handler, options...)
+	customGroup := m.factory.startExistingConsumerGroup(groupId, group, consume, topics, logger, handler, ref, options...)
 	managedGrp.cancelConsume = customGroup.cancel
 
 	// Add the Sarama ConsumerGroup we obtained from the factory to the managed group map,
diff --git a/pkg/common/consumer/consumergroup_offsets_checker.go b/pkg/common/consumer/consumergroup_offsets_checker.go
index e94281d35..bb94a6334 100644
--- a/pkg/common/consumer/consumergroup_offsets_checker.go
+++ b/pkg/common/consumer/consumergroup_offsets_checker.go
@@ -8,6 +8,7 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/util/wait"
+
 	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
 )
 
@@ -18,7 +19,7 @@ const (
 
 // wrapper functions for the Sarama functions, to facilitate unit testing
 var newSaramaClient = sarama.NewClient
-var newSaramaClusterAdmin = sarama.NewClusterAdmin
+var newClusterAdminFromClient = sarama.NewClusterAdminFromClient
 
 type ConsumerGroupOffsetsChecker interface {
 	WaitForOffsetsInitialization(ctx context.Context, groupID string, topics []string, logger *zap.SugaredLogger, addrs []string, config *sarama.Config) error
@@ -44,7 +45,7 @@ func (k *KafkaConsumerGroupOffsetsChecker) WaitForOffsetsInitialization(ctx cont
 	}
 	defer client.Close()
 
-	clusterAdmin, err := newSaramaClusterAdmin(addrs, config)
+	clusterAdmin, err := newClusterAdminFromClient(client)
 	if err != nil {
 		logger.Errorw("unable to create Kafka cluster admin client", zap.Any("topics", topics), zap.String("groupId", groupID), zap.Error(err))
 		return err
@@ -63,10 +64,7 @@ func (k *KafkaConsumerGroupOffsetsChecker) WaitForOffsetsInitialization(ctx cont
 			return false, fmt.Errorf("error checking if offsets are initialized. stopping trying. %w", err)
 		}
 	}
-	pollCtx, pollCtxCancel := context.WithTimeout(ctx, OffsetCheckRetryTimeout)
-	err = wait.PollUntil(OffsetCheckRetryInterval, check, pollCtx.Done())
-	defer pollCtxCancel()
-
+	err = wait.PollImmediate(OffsetCheckRetryInterval, OffsetCheckRetryTimeout, check)
 	if err != nil {
 		return fmt.Errorf("failed to check if offsets are initialized %w", err)
 	}
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index 6017d3b8b..a6934e21f 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -27,6 +27,7 @@ import (
 	"time"
 
 	"golang.org/x/time/rate"
+	"k8s.io/apimachinery/pkg/types"
 	ctrl "knative.dev/control-protocol/pkg"
 	ctrlnetwork "knative.dev/control-protocol/pkg/network"
 
@@ -128,12 +129,13 @@ func (a *Adapter) Start(ctx context.Context) (err error) {
 	a.saramaConfig = config
 
 	options := []consumer.SaramaConsumerHandlerOption{consumer.WithSaramaConsumerLifecycleListener(a)}
-	consumerGroupFactory := consumer.NewConsumerGroupFactory(addrs, config, &consumer.NoopConsumerGroupOffsetsChecker{})
+	consumerGroupFactory := consumer.NewConsumerGroupFactory(addrs, config, &consumer.NoopConsumerGroupOffsetsChecker{}, func(ref types.NamespacedName) {})
 	group, err := consumerGroupFactory.StartConsumerGroup(
 		ctx,
 		a.config.ConsumerGroup,
 		a.config.Topics,
 		a,
+		types.NamespacedName{Namespace: a.config.Namespace, Name: a.config.Name},
 		options...,
 	)
 	if err != nil {


Test patch:
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go b/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
index b58f64e39..a8460bf64 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
@@ -33,6 +33,7 @@ import (
 	"github.com/cloudevents/sdk-go/v2/test"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/types"
 	"knative.dev/eventing/pkg/channel/fanout"
 	"knative.dev/eventing/pkg/kncloudevents"
 	"knative.dev/pkg/apis"
@@ -82,7 +83,7 @@ func TestDispatcher(t *testing.T) {
 	}
 
 	// Create the dispatcher. At this point, if Kafka is not up, this thing fails
-	dispatcher, err := NewDispatcher(context.Background(), &dispatcherArgs)
+	dispatcher, err := NewDispatcher(context.Background(), &dispatcherArgs, func(ref types.NamespacedName) {})
 	if err != nil {
 		t.Skipf("no dispatcher: %v", err)
 	}
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher_test.go b/pkg/channel/consolidated/dispatcher/dispatcher_test.go
index 5245c3d88..69615e6da 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher_test.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher_test.go
@@ -50,7 +50,7 @@ type mockKafkaConsumerFactory struct {
 	createErr bool
 }
 
-func (c mockKafkaConsumerFactory) StartConsumerGroup(ctx context.Context, groupID string, topics []string, handler consumer.KafkaConsumerHandler, options ...consumer.SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error) {
+func (c mockKafkaConsumerFactory) StartConsumerGroup(ctx context.Context, groupID string, topics []string, handler consumer.KafkaConsumerHandler, ref types.NamespacedName, options ...consumer.SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error) {
 	if c.createErr {
 		return nil, errors.New("error creating consumer")
 	}
@@ -519,7 +519,7 @@ func TestNewDispatcher(t *testing.T) {
 		Brokers:   []string{"localhost:10000"},
 		TopicFunc: utils.TopicName,
 	}
-	_, err := NewDispatcher(context.TODO(), args)
+	_, err := NewDispatcher(context.TODO(), args, func(ref types.NamespacedName) {})
 	if err == nil {
 		t.Errorf("Expected error want %s, got %s", "message receiver is not set", err)
 	}
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
index 870c5440b..ef8080775 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"k8s.io/apimachinery/pkg/types"
 	commonenv "knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
 
 	"github.com/stretchr/testify/assert"
@@ -214,7 +215,7 @@ func NewMockDispatcher(t *testing.T) MockDispatcher {
 func (m MockDispatcher) Shutdown() {
 }
 
-func (m MockDispatcher) UpdateSubscriptions(_ context.Context, _ []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error {
+func (m MockDispatcher) UpdateSubscriptions(_ context.Context, ref types.NamespacedName, _ []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error {
 	return nil
 }
 
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
index 0bec5fcc9..c765d518a 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
@@ -338,7 +338,7 @@ func TestUpdateSubscriptions(t *testing.T) {
 			}
 
 			for _, id := range testCase.expectStarted {
-				mockManager.On("StartConsumerGroup", mock.Anything, "kafka."+id, mock.Anything, mock.Anything, mock.Anything).Return(testCase.createErr)
+				mockManager.On("StartConsumerGroup", mock.Anything, "kafka."+id, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(testCase.createErr)
 			}
 			for _, id := range testCase.expectErrors {
 				mockManager.On("Errors", "kafka."+id).Return((<-chan error)(errorSource))
@@ -349,7 +349,7 @@ func TestUpdateSubscriptions(t *testing.T) {
 			}
 
 			// Perform The Test
-			result := dispatcher.UpdateSubscriptions(ctx, testCase.args.subscriberSpecs)
+			result := dispatcher.UpdateSubscriptions(ctx, types.NamespacedName{}, testCase.args.subscriberSpecs)
 
 			close(errorSource)
 
@@ -530,7 +530,7 @@ func createTestDispatcher(t *testing.T, brokers []string, config *sarama.Config)
 	serverHandler.Service.On("SendAndWaitForAck", mock.Anything, mock.Anything).Return(nil)
 
 	// Create The Dispatcher
-	dispatcher := NewDispatcher(dispatcherConfig, serverHandler)
+	dispatcher := NewDispatcher(dispatcherConfig, serverHandler, func(ref types.NamespacedName) {})
 	serverHandler.AssertExpectations(t)
 
 	// Verify State
diff --git a/pkg/common/consumer/consumer_factory_test.go b/pkg/common/consumer/consumer_factory_test.go
index 2db27b3a9..cbc50736e 100644
--- a/pkg/common/consumer/consumer_factory_test.go
+++ b/pkg/common/consumer/consumer_factory_test.go
@@ -24,6 +24,8 @@ import (
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/types"
+
 	controllertesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller/testing"
 	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
 )
@@ -99,13 +101,13 @@ func mockedNewSaramaClient(client *controllertesting.MockClient, mustFail bool)
 	}
 }
 
-func mockedNewSaramaClusterAdmin(clusterAdmin sarama.ClusterAdmin, mustFail bool) func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
+func mockedNewSaramaClusterAdminFromClient(clusterAdmin sarama.ClusterAdmin, mustFail bool) func(client sarama.Client) (sarama.ClusterAdmin, error) {
 	if !mustFail {
-		return func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
+		return func(client sarama.Client) (sarama.ClusterAdmin, error) {
 			return clusterAdmin, nil
 		}
 	} else {
-		return func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
+		return func(client sarama.Client) (sarama.ClusterAdmin, error) {
 			return nil, errors.New("failed")
 		}
 	}
@@ -130,7 +132,7 @@ func TestErrorPropagationCustomConsumerGroup(t *testing.T) {
 	// override some functions
 	newConsumerGroup = mockedNewConsumerGroupFromClient(nil, true, true, false, false)
 	newSaramaClient = mockedNewSaramaClient(client, false)
-	newSaramaClusterAdmin = mockedNewSaramaClusterAdmin(clusterAdmin, false)
+	newClusterAdminFromClient = mockedNewSaramaClusterAdminFromClient(clusterAdmin, false)
 
 	factory := kafkaConsumerGroupFactoryImpl{
 		config:         sarama.NewConfig(),
@@ -138,7 +140,7 @@ func TestErrorPropagationCustomConsumerGroup(t *testing.T) {
 		offsetsChecker: &mockConsumerGroupOffsetsChecker{},
 	}
 
-	consumerGroup, err := factory.StartConsumerGroup(ctx, "bla", []string{}, nil)
+	consumerGroup, err := factory.StartConsumerGroup(ctx, "bla", []string{}, nil, types.NamespacedName{})
 	if err != nil {
 		t.Errorf("Should not throw error %v", err)
 	}
@@ -187,7 +189,7 @@ func TestErrorWhileCreatingNewConsumerGroup(t *testing.T) {
 		addrs:          []string{"b1", "b2"},
 		offsetsChecker: &mockConsumerGroupOffsetsChecker{},
 	}
-	_, err := factory.StartConsumerGroup(ctx, "bla", []string{}, nil)
+	_, err := factory.StartConsumerGroup(ctx, "bla", []string{}, nil, types.NamespacedName{})
 
 	if err == nil || err.Error() != "failed" {
 		t.Errorf("Should contain an error with message failed. Got %v", err)
@@ -203,7 +205,7 @@ func TestErrorWhileNewConsumerGroup(t *testing.T) {
 		addrs:          []string{"b1", "b2"},
 		offsetsChecker: &mockConsumerGroupOffsetsChecker{},
 	}
-	consumerGroup, _ := factory.StartConsumerGroup(ctx, "bla", []string{}, nil)
+	consumerGroup, _ := factory.StartConsumerGroup(ctx, "bla", []string{}, nil, types.NamespacedName{})
 
 	consumerGroup.(*customConsumerGroup).cancel() // Stop the consume loop from spinning after the error is generated
 	err := <-consumerGroup.Errors()
diff --git a/pkg/common/consumer/consumer_manager_test.go b/pkg/common/consumer/consumer_manager_test.go
index ddcd1ce74..63feac90a 100644
--- a/pkg/common/consumer/consumer_manager_test.go
+++ b/pkg/common/consumer/consumer_manager_test.go
@@ -26,6 +26,7 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"k8s.io/apimachinery/pkg/types"
 	ctrl "knative.dev/control-protocol/pkg"
 	ctrlservice "knative.dev/control-protocol/pkg/service"
 	logtesting "knative.dev/pkg/logging/testing"
@@ -37,7 +38,7 @@ import (
 
 func TestNewConsumerGroupManager(t *testing.T) {
 	server := getMockServerHandler()
-	manager := NewConsumerGroupManager(logtesting.TestLogger(t).Desugar(), server, []string{}, &sarama.Config{}, &NoopConsumerGroupOffsetsChecker{})
+	manager := NewConsumerGroupManager(logtesting.TestLogger(t).Desugar(), server, []string{}, &sarama.Config{}, &NoopConsumerGroupOffsetsChecker{}, func(ref types.NamespacedName) {})
 	assert.NotNil(t, manager)
 	assert.NotNil(t, server.Router[commands.StopConsumerGroupOpCode])
 	assert.NotNil(t, server.Router[commands.StartConsumerGroupOpCode])
@@ -108,7 +109,7 @@ func TestStartConsumerGroup(t *testing.T) {
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
 			ctx := context.TODO()
-			manager := NewConsumerGroupManager(logtesting.TestLogger(t).Desugar(), getMockServerHandler(), []string{}, &sarama.Config{}, &NoopConsumerGroupOffsetsChecker{})
+			manager := NewConsumerGroupManager(logtesting.TestLogger(t).Desugar(), getMockServerHandler(), []string{}, &sarama.Config{}, &NoopConsumerGroupOffsetsChecker{}, func(ref types.NamespacedName) {})
 			mockGroup := kafkatesting.NewMockConsumerGroup()
 			newConsumerGroup = func(addrs []string, groupID string, config *sarama.Config) (sarama.ConsumerGroup, error) {
 				if testCase.factoryErr {
@@ -117,7 +118,7 @@ func TestStartConsumerGroup(t *testing.T) {
 				mockGroup.On("Errors").Return(mockGroup.ErrorChan)
 				return mockGroup, nil
 			}
-			err := manager.StartConsumerGroup(ctx, "testid", []string{}, nil, nil)
+			err := manager.StartConsumerGroup(ctx, "testid", []string{}, nil, types.NamespacedName{})
 			assert.Equal(t, testCase.factoryErr, err != nil)
 			time.Sleep(5 * time.Millisecond) // Give the transferErrors routine a chance to call Errors()
 			mockGroup.AssertExpectations(t)
@@ -562,7 +563,7 @@ func getManagerWithMockGroup(t *testing.T, groupId string, factoryErr bool) (Kaf
 		}
 		return kafkatesting.NewMockConsumerGroup(), nil
 	}
-	manager := NewConsumerGroupManager(logtesting.TestLogger(t).Desugar(), serverHandler, []string{}, &sarama.Config{}, &NoopConsumerGroupOffsetsChecker{})
+	manager := NewConsumerGroupManager(logtesting.TestLogger(t).Desugar(), serverHandler, []string{}, &sarama.Config{}, &NoopConsumerGroupOffsetsChecker{}, func(ref types.NamespacedName) {})
 	if groupId != "" {
 		mockGrp, managedGrp := createTestGroup(t)
 		manager.(*kafkaConsumerGroupManagerImpl).groups[groupId] = managedGrp
diff --git a/pkg/common/consumer/testing/mocks.go b/pkg/common/consumer/testing/mocks.go
index 270d8aa89..b0634939d 100644
--- a/pkg/common/consumer/testing/mocks.go
+++ b/pkg/common/consumer/testing/mocks.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/mock"
+	"k8s.io/apimachinery/pkg/types"
 	"knative.dev/eventing-kafka/pkg/common/consumer"
 )
 
@@ -33,8 +34,8 @@ type MockKafkaConsumerGroupFactory struct {
 	mock.Mock
 }
 
-func (c *MockKafkaConsumerGroupFactory) StartConsumerGroup(ctx context.Context, groupId string, topics []string, handler consumer.KafkaConsumerHandler, options ...consumer.SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error) {
-	args := c.Called(ctx, groupId, topics, handler, options)
+func (c *MockKafkaConsumerGroupFactory) StartConsumerGroup(ctx context.Context, groupId string, topics []string, handler consumer.KafkaConsumerHandler, ref types.NamespacedName, options ...consumer.SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error) {
+	args := c.Called(ctx, groupId, topics, handler, ref, options)
 	return args.Get(0).(sarama.ConsumerGroup), args.Error(1)
 }
 
@@ -62,8 +63,8 @@ func (m *MockConsumerGroupManager) Reconfigure(brokers []string, config *sarama.
 }
 
 func (m *MockConsumerGroupManager) StartConsumerGroup(ctx context.Context, groupId string, topics []string,
-	handler consumer.KafkaConsumerHandler, options ...consumer.SaramaConsumerHandlerOption) error {
-	args := m.Called(ctx, groupId, topics, handler, options)
+	handler consumer.KafkaConsumerHandler, channelRef types.NamespacedName, options ...consumer.SaramaConsumerHandlerOption) error {
+	args := m.Called(ctx, groupId, topics, handler, channelRef, options)
 	return args.Error(0)
 }