Repository: knative-sandbox/eventing-kafka
Pull request #929 (2021-10-08 13:24:28)

[0.25] KafkaChannel dispatcher offset checking improvements
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Backport https://github.com/knative-sandbox/eventing-kafka/pull/928
-
-

<!--
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

```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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 | f59eee2c13a2a7d84ec2ce05c59ab3437cd5109f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/929/pull-knative-sandbox-eventing-kafka-unit-tests/1446466531197194240) | true | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | f59eee2c13a2a7d84ec2ce05c59ab3437cd5109f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/929/pull-knative-sandbox-eventing-kafka-go-coverage/1446466531222360064) | false | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/929). [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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/929?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> :exclamation: No coverage uploaded for pull request base (`release-0.25@813d807`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#section-missing-base-commit).
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/929/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/929?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@               Coverage Diff               @@
##             release-0.25     #929   +/-   ##
===============================================
  Coverage                ?   74.07%           
===============================================
  Files                   ?      139           
  Lines                   ?     6215           
  Branches                ?        0           
===============================================
  Hits                    ?     4604           
  Misses                  ?     1381           
  Partials                ?      230           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/929?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/929?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [813d807...f59eee2](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/929?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

pierDipi: All green 
/test pull-knative-sandbox-eventing-kafka-upgrade-tests
pierDipi: All green
/test pull-knative-sandbox-eventing-kafka-upgrade-tests

Patch:
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index 5c593bd48..7760868db 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -153,7 +153,7 @@ func main() {
 		MetricsRegistry: ekConfig.Sarama.Config.MetricRegistry,
 		SaramaConfig:    ekConfig.Sarama.Config,
 	}
-	dispatcher, managerEvents := dispatch.NewDispatcher(dispatcherConfig, controlProtocolServer)
+	dispatcher, managerEvents := dispatch.NewDispatcher(dispatcherConfig, controlProtocolServer, func(ref types.NamespacedName) {})
 
 	// Create KafkaChannel Informer
 	kafkaClient := kafkaclientset.NewForConfigOrDie(k8sConfig)
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 d9c898d5c..5f84f728a 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
@@ -231,7 +231,11 @@ func (r Reconciler) reconcile(ctx context.Context, channel *kafkav1beta1.KafkaCh
 	}
 
 	// Update The ConsumerGroups To Align With Current KafkaChannel Subscribers
-	subscriptions := r.dispatcher.UpdateSubscriptions(ctx, subscribers)
+	channelRef := types.NamespacedName{
+		Namespace: channel.GetNamespace(),
+		Name:      channel.GetName(),
+	}
+	subscriptions := r.dispatcher.UpdateSubscriptions(ctx, channelRef, subscribers)
 
 	// Update The KafkaChannel Subscribable Status Based On ConsumerGroup Creation Status
 	channel.Status.SubscribableStatus = r.createSubscribableStatus(channel.Spec.Subscribers, subscriptions)
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index f2d74c737..c41653434 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -65,7 +65,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) commonconsumer.SubscriberStatusMap
+	UpdateSubscriptions(ctx context.Context, channelRef types.NamespacedName, subscriberSpecs []eventingduck.SubscriberSpec) commonconsumer.SubscriberStatusMap
 }
 
 // DispatcherImpl Is A Struct With Configuration & ConsumerGroup State
@@ -83,9 +83,9 @@ type DispatcherImpl struct {
 var _ Dispatcher = &DispatcherImpl{}
 
 // NewDispatcher Is The Dispatcher Constructor
-func NewDispatcher(dispatcherConfig DispatcherConfig, controlServer controlprotocol.ServerHandler) (Dispatcher, <-chan commonconsumer.ManagerEvent) {
+func NewDispatcher(dispatcherConfig DispatcherConfig, controlServer controlprotocol.ServerHandler, enqueue func(ref types.NamespacedName)) (Dispatcher, <-chan commonconsumer.ManagerEvent) {
 
-	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{
@@ -123,7 +123,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) commonconsumer.SubscriberStatusMap {
+func (d *DispatcherImpl) UpdateSubscriptions(ctx context.Context, channelRef types.NamespacedName, subscriberSpecs []eventingduck.SubscriberSpec) commonconsumer.SubscriberStatusMap {
 
 	if d.SaramaConfig == nil {
 		d.Logger.Error("Dispatcher has no config!")
@@ -151,7 +151,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..1a6833f95 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)
@@ -107,6 +110,8 @@ func (c kafkaConsumerGroupFactoryImpl) startExistingConsumerGroup(
 		if err != nil {
 			logger.Errorw("error while checking if offsets are initialized", zap.Any("topics", topics), zap.String("groupId", groupID), 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 02a2bcc2b..882e4143d 100644
--- a/pkg/common/consumer/consumer_manager.go
+++ b/pkg/common/consumer/consumer_manager.go
@@ -94,7 +94,7 @@ type ManagerEvent struct {
 // KafkaConsumerGroupManager keeps track of Sarama consumer groups and handles messages from control-protocol clients
 type KafkaConsumerGroupManager interface {
 	Reconfigure(brokers []string, config *sarama.Config) *ReconfigureError
-	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
@@ -141,13 +141,13 @@ func (r ReconfigureError) Error() string {
 var _ error = (*ReconfigureError)(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{},
 		eventLock:      sync.Mutex{},
 		offsetsChecker: offsetsChecker,
@@ -264,7 +264,7 @@ func (m *kafkaConsumerGroupManagerImpl) Reconfigure(brokers []string, config *sa
 
 // 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))
@@ -286,7 +286,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 := createManagedGroup(ctx, m.logger, group, cancel, 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 15b69667c..d09d29566 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 2c65dab8b..2e0ad8a8e 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
@@ -317,7 +317,7 @@ func TestAllCases(t *testing.T) {
 		status consumer.SubscriberStatusMap,
 	) controller.Reconciler {
 		mockDispatcher := &MockDispatcher{}
-		mockDispatcher.On("UpdateSubscriptions", mock.Anything).Return(status)
+		mockDispatcher.On("UpdateSubscriptions", mock.Anything, mock.Anything, mock.Anything).Return(status)
 		return &Reconciler{
 			logger:               logtesting.TestLogger(t).Desugar(),
 			channelKey:           kcKey,
@@ -365,8 +365,8 @@ func (m *MockDispatcher) Shutdown() {
 	m.Called()
 }
 
-func (m *MockDispatcher) UpdateSubscriptions(_ context.Context, subscriberSpecs []eventingduck.SubscriberSpec) consumer.SubscriberStatusMap {
-	args := m.Called(subscriberSpecs)
+func (m *MockDispatcher) UpdateSubscriptions(ctx context.Context, ref types.NamespacedName, subscriberSpecs []eventingduck.SubscriberSpec) consumer.SubscriberStatusMap {
+	args := m.Called(ctx, ref, subscriberSpecs)
 	return args.Get(0).(consumer.SubscriberStatusMap)
 }
 
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
index f529b63a5..cd9ae9d80 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
@@ -365,7 +365,7 @@ func TestUpdateSubscriptions(t *testing.T) {
 			if !testCase.wantFailure {
 				mockManager.On("ClearNotifications").Return()
 				for _, id := range testCase.expectStarted {
-					mockManager.On("StartConsumerGroup", mock.Anything, "kafka."+id, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(testCase.createErr)
+					mockManager.On("StartConsumerGroup", mock.Anything, "kafka."+id, mock.Anything, 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))
@@ -380,7 +380,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)
 
@@ -572,7 +572,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, events := NewDispatcher(dispatcherConfig, serverHandler)
+	dispatcher, events := NewDispatcher(dispatcherConfig, serverHandler, func(ref types.NamespacedName) {})
 	assert.NotNil(t, events)
 	serverHandler.AssertExpectations(t)
 
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 b5a341de7..dd1e45e7a 100644
--- a/pkg/common/consumer/consumer_manager_test.go
+++ b/pkg/common/consumer/consumer_manager_test.go
@@ -27,6 +27,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"
@@ -38,7 +39,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])
@@ -156,7 +157,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 {
@@ -165,7 +166,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)
@@ -621,7 +622,7 @@ func getManagerWithMockGroup(t *testing.T, groupId string, factoryErr bool) (Kaf
 		mockGroup.On("Errors").Return(make(chan error))
 		return mockGroup, 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 != "" {
 		mockGroup, managedGrp := createMockAndManagedGroups(t)
 		manager.(*kafkaConsumerGroupManagerImpl).groups[groupId] = managedGrp
diff --git a/pkg/common/consumer/testing/mocks.go b/pkg/common/consumer/testing/mocks.go
index 428d66f13..26fdec644 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)
 }
 
@@ -61,8 +62,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 {
-	return m.Called(ctx, groupId, topics, handler, options).Error(0)
+	handler consumer.KafkaConsumerHandler, channelRef types.NamespacedName, options ...consumer.SaramaConsumerHandlerOption) error {
+	return m.Called(ctx, groupId, topics, handler, channelRef, options).Error(0)
 }
 
 func (m *MockConsumerGroupManager) CloseConsumerGroup(groupId string) error {