Repository: knative-sandbox/eventing-kafka
Pull request #739 (2021-06-24 17:56:56)

[Source] Optimize InitOffsets

<!-- Please include the 'why' behind your changes if no issue exists -->
InitOffsets can take a long time (several seconds, sometimes minutes), when the number of partitions is high (> 500), due to Sarama inability to batch OffsetFetch requests. 


## Proposed Changes

- batch OffsetFetch requests, one per leading broker. 
- 
-

<!--
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: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/739#" title="Author self-approved">lionelvillard</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/main/OWNERS)~~ [lionelvillard]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/kafka/sarama/batch_get_offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/739/pull-knative-sandbox-eventing-kafka-go-coverage/1408121986965573632/artifacts/line-cov.html#file0) | Do not exist | 85.7% | 
[pkg/source/client/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/739/pull-knative-sandbox-eventing-kafka-go-coverage/1408121986965573632/artifacts/line-cov.html#file1) | 81.8% | 79.5% | -2.3

knative-prow-robot: @lionelvillard: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 7dc1ee1475bf8fa153adc4ca736b064fb19fb934 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/739/pull-knative-sandbox-eventing-kafka-unit-tests/1408121986923630592) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

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

<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/739?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#739](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (7dc1ee1) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/000dd9c4955f4b3c3e2f71b35d30138c7d1b9770?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (000dd9c) will **decrease** coverage by `1.52%`.
> The diff coverage is `51.21%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739/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/739?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@            Coverage Diff             @@
##             main     #739      +/-   ##
==========================================
- Coverage   75.29%   73.76%   -1.53%     
==========================================
  Files         132      134       +2     
  Lines        5785     5909     +124     
==========================================
+ Hits         4356     4359       +3     
- Misses       1208     1321     +113     
- Partials      221      229       +8     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...onsolidated/dispatcher/consumer\_message\_handler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvY29uc3VtZXJfbWVzc2FnZV9oYW5kbGVyLmdv) | `0.00% <0.00%> (ø)` | |
| [pkg/source/adapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9hZGFwdGVyL2FkYXB0ZXIuZ28=) | `58.42% <ø> (-1.08%)` | :arrow_down: |
| [pkg/source/client/client.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9jbGllbnQvY2xpZW50Lmdv) | `15.66% <0.00%> (-19.48%)` | :arrow_down: |
| [pkg/source/reconciler/source/kafkasource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9yZWNvbmNpbGVyL3NvdXJjZS9rYWZrYXNvdXJjZS5nbw==) | `0.00% <0.00%> (ø)` | |
| [pkg/source/mtadapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvYWRhcHRlci5nbw==) | `47.82% <22.72%> (-27.36%)` | :arrow_down: |
| [pkg/apis/sources/v1beta1/kafka\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMvc291cmNlcy92MWJldGExL2thZmthX2xpZmVjeWNsZS5nbw==) | `53.33% <50.00%> (-0.52%)` | :arrow_down: |
| [pkg/channel/consolidated/utils/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3V0aWxzL3V0aWwuZ28=) | `95.91% <50.00%> (-4.09%)` | :arrow_down: |
| [pkg/source/client/offsets.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9jbGllbnQvb2Zmc2V0cy5nbw==) | `55.55% <55.55%> (ø)` | |
| [pkg/common/kafka/sarama/batch\_get\_offsets.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9rYWZrYS9zYXJhbWEvYmF0Y2hfZ2V0X29mZnNldHMuZ28=) | `65.21% <65.21%> (ø)` | |
| [pkg/channel/consolidated/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvZGlzcGF0Y2hlci5nbw==) | `61.58% <66.66%> (+0.10%)` | :arrow_up: |
| ... and [10 more](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739?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/739?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [d217e99...7dc1ee1](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/739?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).

steven0711dong: * If my understanding is correct, instead of calling the GetOffset function partition number equivalent of times, we are only making broker number equivalent of times to get all the offsets? 
* When we construct the OffsetRequest, don't we need to fill the blocks as well? The way we do it is very simple by only filling out the version. 
* Lastly, have you tested this PR somewhere? 
lionelvillard: > * If my understanding is correct, instead of calling the GetOffset function partition number equivalent of times, we are only making broker number equivalent of times to get all the offsets?

Correct.

> * When we construct the OffsetRequest, don't we need to fill the blocks as well? The way we do it is very simple by only filling out the version.

See `request.AddBlock`. 

> * Lastly, have you tested this PR somewhere?

Besides unit testing and local testing? Existing e2e tests run this code and there is an open [PR](https://github.com/knative-sandbox/eventing-kafka/pull/636) to specifically test there is no missing events after the source is ready. 


knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/kafka/sarama/batch_get_offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/739/pull-knative-sandbox-eventing-kafka-go-coverage/1413132428616142848/artifacts/line-cov.html#file0) | Do not exist | 85.7% | 
[pkg/source/client/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/739/pull-knative-sandbox-eventing-kafka-go-coverage/1413132428616142848/artifacts/line-cov.html#file1) | 81.8% | 79.5% | -2.3


Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 24c126c6f..566c2b992 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -5,7 +5,7 @@ aliases:
   api-core-wg-leads:
   - dprotaso
   async-component-approvers:
-  - beemarie
+  - bvennam
   - julz
   - maximilien
   autoscaling-wg-leads:
@@ -21,11 +21,9 @@ aliases:
   - devguyio
   - lionelvillard
   - matzew
-  - slinkydeveloper
   - vaikas
   delivery-wg-leads:
   - matzew
-  - slinkydeveloper
   discovery-approvers:
   - lberk
   - n3wscott
@@ -63,7 +61,6 @@ aliases:
   - travis-minke-sap
   eventing-kafka-broker-approvers:
   - pierDipi
-  - slinkydeveloper
   eventing-kafka-mtsource-approvers:
   - steven0711dong
   eventing-natss-approvers:
@@ -87,7 +84,6 @@ aliases:
   - lionelvillard
   - matzew
   - n3wscott
-  - slinkydeveloper
   - vaikas
   homebrew-kn-plugins-approvers:
   - dsimansk
@@ -153,12 +149,12 @@ aliases:
   - k4leung4
   - lionelvillard
   - markusthoemmes
+  - matzew
   - mikehelmick
   - n3wscott
   - nak3
   - navidshaikh
   - rhuss
-  - slinkydeveloper
   - tcnghia
   - vagababov
   - vaikas
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index 19ecd000e..05d14f982 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -26,6 +26,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	"knative.dev/eventing/pkg/kncloudevents"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
 	kncontroller "knative.dev/pkg/controller"
@@ -45,16 +46,17 @@ import (
 	kafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	"knative.dev/eventing-kafka/pkg/client/informers/externalversions"
 	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
+	commonconsumer "knative.dev/eventing-kafka/pkg/common/consumer"
 	"knative.dev/eventing-kafka/pkg/common/controlprotocol"
 	"knative.dev/eventing-kafka/pkg/common/kafka/sarama"
 	"knative.dev/eventing-kafka/pkg/common/metrics"
-	"knative.dev/eventing/pkg/kncloudevents"
 )
 
 // Variables
 var (
-	logger     *zap.Logger
-	dispatcher dispatch.Dispatcher
+	logger        *zap.Logger
+	dispatcher    dispatch.Dispatcher
+	managerEvents <-chan commonconsumer.ManagerEvent
 )
 
 // The Main Function (Go Command)
@@ -157,7 +159,7 @@ func main() {
 		MetricsRegistry: ekConfig.Sarama.Config.MetricRegistry,
 		SaramaConfig:    ekConfig.Sarama.Config,
 	}
-	dispatcher = dispatch.NewDispatcher(dispatcherConfig, controlProtocolServer)
+	dispatcher, managerEvents = dispatch.NewDispatcher(dispatcherConfig, controlProtocolServer)
 
 	// Watch The Secret For Changes
 	err = distributedcommonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, environment.ResyncPeriod, secretObserver)
@@ -182,6 +184,7 @@ func main() {
 			k8sClient,
 			kafkaClient,
 			ctx.Done(),
+			managerEvents,
 		),
 	}
 
diff --git a/pkg/channel/distributed/dispatcher/constants/constants.go b/pkg/channel/distributed/dispatcher/constants/constants.go
index 280f936c9..fa48ecb1b 100644
--- a/pkg/channel/distributed/dispatcher/constants/constants.go
+++ b/pkg/channel/distributed/dispatcher/constants/constants.go
@@ -23,4 +23,7 @@ const (
 	MetricsInterval = 5 * time.Second
 
 	Component = "eventing-kafka-channel-dispatcher"
+
+	// GroupStoppedMessage is the message that will be in a subscriber's status when a group is stopped ("paused")
+	GroupStoppedMessage = "consumer group is stopped"
 )
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
index 93f7e8431..a319de7e1 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
@@ -25,20 +25,24 @@ 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"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+
 	kafkav1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/dispatcher"
 	"knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	"knative.dev/eventing-kafka/pkg/client/clientset/versioned/scheme"
 	informers "knative.dev/eventing-kafka/pkg/client/informers/externalversions/messaging/v1beta1"
 	listers "knative.dev/eventing-kafka/pkg/client/listers/messaging/v1beta1"
-	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
-	"knative.dev/pkg/controller"
-	"knative.dev/pkg/logging"
+	commonconsumer "knative.dev/eventing-kafka/pkg/common/consumer"
 )
 
 const (
@@ -75,6 +79,7 @@ func NewController(
 	kubeClient kubernetes.Interface,
 	kafkaClientSet versioned.Interface,
 	stopChannel <-chan struct{},
+	managerEvents <-chan commonconsumer.ManagerEvent,
 ) *controller.Impl {
 
 	reconciler := &Reconciler{
@@ -106,10 +111,51 @@ func NewController(
 			w.Stop()
 		}
 	}()
-
+	if err := reconciler.processManagerEvents(managerEvents); err != nil {
+		logger.Warn("Could not begin processing events from the Consumer Group Manager", zap.Error(err))
+	}
 	return reconciler.impl
 }
 
+// processManagerEvents will listen on the channel provided by the KafkaConsumerGroupManager for events
+// related to changes in the status of a managed ConsumerGroup.  This function is non-blocking, and the
+// internal goroutine will exit when the channel is closed.
+func (r Reconciler) processManagerEvents(events <-chan commonconsumer.ManagerEvent) error {
+	// If the events channel is nil, there's no point listening to it; it will just block forever
+	if events == nil {
+		return fmt.Errorf("no event channel provided")
+	}
+	// Find the namespace and name of the KafkaChannel this dispatcher is monitoring
+	namespace, name, err := cache.SplitMetaNamespaceKey(r.channelKey)
+	if err != nil {
+		return fmt.Errorf("invalid resource key")
+	}
+	key := types.NamespacedName{Namespace: namespace, Name: name}
+	go func() {
+		for event := range events {
+			groupLogger := r.logger.With(zap.String("groupId", event.GroupId))
+			// Stopping or Starting a consumer group requires a reconciliation in order to adjust the status block
+			// of the KafkaChannel - EnqueueKey will do that
+			switch event.Event {
+			case commonconsumer.GroupStopped:
+				groupLogger.Debug("Processing GroupStopped Event From Consumer Group Manager")
+				r.impl.EnqueueKey(key)
+			case commonconsumer.GroupStarted:
+				groupLogger.Debug("Processing GroupStarted Event From Consumer Group Manager")
+				r.impl.EnqueueKey(key)
+			case commonconsumer.GroupCreated:
+				groupLogger.Debug("Processing GroupCreated Event From Consumer Group Manager")
+			case commonconsumer.GroupClosed:
+				groupLogger.Debug("Processing GroupClosed Event From Consumer Group Manager")
+			default:
+				groupLogger.Warn("Received Unexpected Event From Consumer Group Manager", zap.Int("Event", int(event.Event)))
+			}
+		}
+		r.logger.Debug("Manager event channel closed")
+	}()
+	return nil
+}
+
 func (r Reconciler) Reconcile(ctx context.Context, key string) error {
 
 	r.logger.Info("Reconcile", zap.String("key", key))
@@ -180,14 +226,14 @@ func (r Reconciler) reconcile(channel *kafkav1beta1.KafkaChannel) error {
 	}
 
 	// Update The ConsumerGroups To Align With Current KafkaChannel Subscribers
-	failedSubscriptions := r.dispatcher.UpdateSubscriptions(subscribers)
+	subscriptions := r.dispatcher.UpdateSubscriptions(subscribers)
 
 	// Update The KafkaChannel Subscribable Status Based On ConsumerGroup Creation Status
-	channel.Status.SubscribableStatus = r.createSubscribableStatus(channel.Spec.Subscribers, failedSubscriptions)
+	channel.Status.SubscribableStatus = r.createSubscribableStatus(channel.Spec.Subscribers, subscriptions)
 
 	// Log Failed Subscriptions & Return Error
-	if len(failedSubscriptions) > 0 {
-		r.logger.Error("Failed To Subscribe Kafka Subscriptions", zap.Int("Count", len(failedSubscriptions)))
+	if failed := subscriptions.FailedCount(); failed > 0 {
+		r.logger.Error("Failed To Subscribe Kafka Subscriptions", zap.Int("Count", failed))
 		return fmt.Errorf("some kafka subscribers failed to subscribe")
 	}
 
@@ -196,7 +242,7 @@ func (r Reconciler) reconcile(channel *kafkav1beta1.KafkaChannel) error {
 }
 
 // Create The SubscribableStatus Block Based On The Updated Subscriptions
-func (r *Reconciler) createSubscribableStatus(subscribers []eventingduck.SubscriberSpec, failedSubscriptions map[eventingduck.SubscriberSpec]error) eventingduck.SubscribableStatus {
+func (r *Reconciler) createSubscribableStatus(subscribers []eventingduck.SubscriberSpec, subscriptions commonconsumer.SubscriberStatusMap) eventingduck.SubscribableStatus {
 
 	subscriberStatus := make([]eventingduck.SubscriberStatus, 0)
 
@@ -206,10 +252,19 @@ func (r *Reconciler) createSubscribableStatus(subscribers []eventingduck.Subscri
 			ObservedGeneration: subscriber.Generation,
 			Ready:              corev1.ConditionTrue,
 		}
-		if err, ok := failedSubscriptions[subscriber]; ok {
+		// If the UID isn't in the subscription map, the zero-value of the subscriptionStatus will have a nil Error
+		// and Stopped will be false.
+		subscriptionStatus := subscriptions[subscriber.UID]
+		if subscriptionStatus.Error != nil {
 			status.Ready = corev1.ConditionFalse
-			status.Message = err.Error()
+			status.Message = subscriptionStatus.Error.Error()
+		} else if subscriptionStatus.Stopped {
+			// A stopped group isn't an "error" but it does represent a group that isn't "Ready" as far
+			// as subscriber status goes.
+			status.Ready = corev1.ConditionFalse
+			status.Message = constants.GroupStoppedMessage
 		}
+
 		subscriberStatus = append(subscriberStatus, status)
 	}
 
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index 07c780ac3..a10ac2d11 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(subscriberSpecs []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error
+	UpdateSubscriptions(subscriberSpecs []eventingduck.SubscriberSpec) commonconsumer.SubscriberStatusMap
 }
 
 // DispatcherImpl Is A Struct With Configuration & ConsumerGroup State
@@ -84,7 +84,7 @@ 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) (Dispatcher, <-chan commonconsumer.ManagerEvent) {
 
 	consumerGroupManager := commonconsumer.NewConsumerGroupManager(dispatcherConfig.Logger, controlServer, dispatcherConfig.Brokers, dispatcherConfig.SaramaConfig)
 
@@ -102,7 +102,7 @@ func NewDispatcher(dispatcherConfig DispatcherConfig, controlServer controlproto
 	dispatcher.ObserveMetrics(dispatcherconstants.MetricsInterval)
 
 	// Return The DispatcherImpl
-	return dispatcher
+	return dispatcher, consumerGroupManager.GetNotificationChannel()
 }
 
 // Shutdown The Dispatcher
@@ -118,10 +118,13 @@ func (d *DispatcherImpl) Shutdown() {
 	for _, subscriber := range d.subscribers {
 		d.closeConsumerGroup(subscriber)
 	}
+
+	// Close the Consumer Group Manager notification channels
+	d.consumerMgr.ClearNotifications()
 }
 
 // UpdateSubscriptions manages the Dispatcher's Subscriptions to align with new state
-func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error {
+func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.SubscriberSpec) commonconsumer.SubscriberStatusMap {
 
 	if d.SaramaConfig == nil {
 		d.Logger.Error("Dispatcher has no config!")
@@ -129,8 +132,7 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 	}
 
 	// Maps For Tracking Subscriber State
-	activeSubscriptions := make(map[types.UID]bool)
-	failedSubscriptions := make(map[eventingduck.SubscriberSpec]error)
+	subscriptions := make(commonconsumer.SubscriberStatusMap)
 
 	// Thread Safe ;)
 	d.consumerUpdateLock.Lock()
@@ -139,12 +141,12 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 	// Loop Over All All The Specified Subscribers
 	for _, subscriberSpec := range subscriberSpecs {
 
+		// Format The GroupId For The Specified Subscriber
+		groupId := commonkafkautil.GroupId(string(subscriberSpec.UID))
+
 		// If The Subscriber Wrapper For The SubscriberSpec Does Not Exist Then Create One
 		if _, ok := d.subscribers[subscriberSpec.UID]; !ok {
 
-			// Format The GroupId For The Specified Subscriber
-			groupId := commonkafkautil.GroupId(string(subscriberSpec.UID))
-
 			// Create A ConsumerGroup Logger
 			logger := d.Logger.With(zap.String("GroupId", groupId))
 
@@ -155,8 +157,7 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 
 				// Log & Return Failure
 				logger.Error("Failed To Create ConsumerGroup", zap.Error(err))
-				failedSubscriptions[subscriberSpec] = err
-
+				subscriptions[subscriberSpec.UID] = commonconsumer.SubscriberStatus{Error: err}
 			} else {
 
 				// Create A New SubscriberWrapper With The ConsumerGroup
@@ -165,21 +166,27 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 				// Asynchronously Process ConsumerGroup's Error Channel
 				go func() {
 					logger.Info("ConsumerGroup Error Processing Initiated")
-					for err := range d.consumerMgr.Errors(subscriber.GroupId) { // Closing ConsumerGroup Will Break Out Of This
-						logger.Error("ConsumerGroup Error", zap.Error(err))
+					for groupErr := range d.consumerMgr.Errors(subscriber.GroupId) { // Closing ConsumerGroup Will Break Out Of This
+						logger.Error("ConsumerGroup Error", zap.Error(groupErr))
 					}
 					logger.Info("ConsumerGroup Error Processing Terminated")
 				}()
 
 				// Track The New SubscriberWrapper For The SubscriberSpec As Active
 				d.subscribers[subscriberSpec.UID] = subscriber
-				activeSubscriptions[subscriberSpec.UID] = true
+				subscriptions[subscriberSpec.UID] = commonconsumer.SubscriberStatus{}
 			}
 
 		} else {
-
 			// Otherwise Just Add To List Of Active Subscribers
-			activeSubscriptions[subscriberSpec.UID] = true
+			subscriptions[subscriberSpec.UID] = commonconsumer.SubscriberStatus{}
+
+			// If the group is stopped, it's still active but the reconciler needs to know about it in order
+			// to not treat it as a failure (which would re-create the group, effectively un-stopping it)
+			if d.consumerMgr.IsStopped(groupId) {
+				d.Logger.Debug("Adding Stopped ConsumerGroup To Stopped Map", zap.String("GroupId", groupId))
+				subscriptions[subscriberSpec.UID] = commonconsumer.SubscriberStatus{Stopped: true}
+			}
 		}
 	}
 
@@ -187,9 +194,10 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 	// if necessary without going through the inactive subscribers again.
 	d.SubscriberSpecs = []eventingduck.SubscriberSpec{}
 
-	// Close ConsumerGroups For Removed Subscriptions (In Map But No Longer Active)
+	// Close ConsumerGroups For Removed/Failed Subscriptions (In Map But No Longer Active)
 	for _, subscriber := range d.subscribers {
-		if !activeSubscriptions[subscriber.UID] {
+		subscription, ok := subscriptions[subscriber.UID]
+		if !ok || subscription.Error != nil {
 			d.closeConsumerGroup(subscriber)
 		} else {
 			d.SubscriberSpecs = append(d.SubscriberSpecs, subscriber.SubscriberSpec)
@@ -197,7 +205,7 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 	}
 
 	// Return Any Failed Subscriber Errors
-	return failedSubscriptions
+	return subscriptions
 }
 
 // closeConsumerGroup closes the ConsumerGroup associated with a single Subscriber
diff --git a/pkg/common/consumer/consumer_manager.go b/pkg/common/consumer/consumer_manager.go
index 088def87a..8279b31b8 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/eventing-kafka/pkg/common/controlprotocol"
@@ -52,6 +53,43 @@ const (
 	internalToken      = "internal-token"
 )
 
+// SubscriberStatus keeps track of the difference between active, failed, and stopped subscribers
+type SubscriberStatus struct {
+	Stopped bool  // A stopped subscriber is active but suspended ("paused") and is not processing events
+	Error   error // A subscriber with a non-nil error has failed
+}
+
+// SubscriberStatusMap defines the map type which holds a collection of Subscribers by UID and their status
+type SubscriberStatusMap map[types.UID]SubscriberStatus
+
+// FailedCount returns the count of subscribers represented by this map which have an errors associated with them
+func (s SubscriberStatusMap) FailedCount() int {
+	failed := 0
+	for _, status := range s {
+		if status.Error != nil {
+			failed++
+		}
+	}
+	return failed
+}
+
+// EventIndex is the type of Event used when sending ManagerEvent structs via the notifyChannels list
+type EventIndex int
+
+// Events
+const (
+	GroupCreated EventIndex = iota + 1
+	GroupStopped
+	GroupStarted
+	GroupClosed
+)
+
+// ManagerEvent is the struct used by the notification channel
+type ManagerEvent struct {
+	Event   EventIndex
+	GroupId string
+}
+
 // KafkaConsumerGroupManager keeps track of Sarama consumer groups and handles messages from control-protocol clients
 type KafkaConsumerGroupManager interface {
 	Reconfigure(brokers []string, config *sarama.Config) error
@@ -59,6 +97,9 @@ type KafkaConsumerGroupManager interface {
 	CloseConsumerGroup(groupId string) error
 	Errors(groupId string) <-chan error
 	IsManaged(groupId string) bool
+	IsStopped(groupId string) bool
+	GetNotificationChannel() <-chan ManagerEvent
+	ClearNotifications()
 }
 
 // groupMap is a mapping of GroupIDs to managed Consumer Group interfaces
@@ -67,11 +108,13 @@ type groupMap map[string]managedGroup
 // kafkaConsumerGroupManagerImpl is the primary implementation of a KafkaConsumerGroupManager, which
 // handles control protocol messages and stopping/starting ("pausing/resuming") of ConsumerGroups.
 type kafkaConsumerGroupManagerImpl struct {
-	logger    *zap.Logger
-	server    controlprotocol.ServerHandler
-	factory   *kafkaConsumerGroupFactoryImpl
-	groups    groupMap
-	groupLock sync.RWMutex // Synchronizes write access to the groupMap
+	logger         *zap.Logger
+	server         controlprotocol.ServerHandler
+	factory        *kafkaConsumerGroupFactoryImpl
+	groups         groupMap
+	groupLock      sync.RWMutex // Synchronizes write access to the groupMap
+	notifyChannels []chan ManagerEvent
+	eventLock      sync.Mutex
 }
 
 // Verify that the kafkaConsumerGroupManagerImpl satisfies the KafkaConsumerGroupManager interface
@@ -86,6 +129,7 @@ func NewConsumerGroupManager(logger *zap.Logger, serverHandler controlprotocol.S
 		groups:    make(groupMap),
 		factory:   &kafkaConsumerGroupFactoryImpl{addrs: brokers, config: config},
 		groupLock: sync.RWMutex{},
+		eventLock: sync.Mutex{},
 	}
 
 	logger.Info("Registering Consumer Group Manager Control-Protocol Handlers")
@@ -111,6 +155,38 @@ func NewConsumerGroupManager(logger *zap.Logger, serverHandler controlprotocol.S
 	return manager
 }
 
+// GetNotificationChannel creates a new ManagerEvent channel and returns it.  Manager events will be
+// broadcast to all channels created this way.
+func (m *kafkaConsumerGroupManagerImpl) GetNotificationChannel() <-chan ManagerEvent {
+	eventChan := make(chan ManagerEvent)
+	m.eventLock.Lock()
+	m.notifyChannels = append(m.notifyChannels, eventChan)
+	m.eventLock.Unlock()
+	return eventChan
+}
+
+// ClearNotifications closes and removes all channels from the notification list
+func (m *kafkaConsumerGroupManagerImpl) ClearNotifications() {
+	for _, eventChan := range m.notifyChannels {
+		close(eventChan)
+	}
+	m.eventLock.Lock()
+	m.notifyChannels = nil
+	m.eventLock.Unlock()
+}
+
+// notify will send the given ManagerEvent to all channels in the notifyChannels list
+func (m *kafkaConsumerGroupManagerImpl) notify(event ManagerEvent) {
+	m.logger.Debug("Notifying channels", zap.Int("count", len(m.notifyChannels)), zap.Any("event", event))
+	for _, eventChan := range m.notifyChannels {
+		// Don't block if the receiver isn't listening
+		select {
+		case eventChan <- event:
+		default:
+		}
+	}
+}
+
 // Reconfigure will incorporate a new set of brokers and Sarama config settings into the manager
 // without requiring a new control-protocol server or losing the current map of managed groups.
 // It will stop and start all of the managed groups in the group map.
@@ -172,6 +248,7 @@ func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(groupId string, topic
 	// Add the Sarama ConsumerGroup we obtained from the factory to the managed group map,
 	// so that it can be stopped and started via control-protocol messages.
 	m.setGroup(groupId, managedGrp)
+	m.notify(ManagerEvent{Event: GroupCreated, GroupId: groupId})
 	return nil
 }
 
@@ -193,6 +270,7 @@ func (m *kafkaConsumerGroupManagerImpl) CloseConsumerGroup(groupId string) error
 
 	// Remove this groupId from the map so that manager functions may not be called on it
 	m.removeGroup(groupId)
+	m.notify(ManagerEvent{Event: GroupClosed, GroupId: groupId})
 
 	return nil
 }
@@ -213,6 +291,16 @@ func (m *kafkaConsumerGroupManagerImpl) IsManaged(groupId string) bool {
 	return m.getGroup(groupId) != nil
 }
 
+// IsStopped returns true if the given groupId corresponds to a stopped ConsumerGroup
+func (m *kafkaConsumerGroupManagerImpl) IsStopped(groupId string) bool {
+	group := m.getGroup(groupId)
+	if group == nil {
+		return false // If it's not a managed group, it can't be stopped
+	}
+
+	return group.isStopped()
+}
+
 // Consume calls the Consume method of a managed consumer group, using a loop to call it again if that
 // group is restarted by the manager.  If the Consume call is terminated by some other mechanism, the
 // result will be returned to the caller.
@@ -253,6 +341,7 @@ func (m *kafkaConsumerGroupManagerImpl) stopConsumerGroup(lock *commands.Command
 		groupLogger.Error("Failed to unlock consumer group after stopping", zap.Error(err))
 		return err
 	}
+	m.notify(ManagerEvent{Event: GroupStopped, GroupId: groupId})
 	return nil
 }
 
@@ -289,6 +378,7 @@ func (m *kafkaConsumerGroupManagerImpl) startConsumerGroup(lock *commands.Comman
 		groupLogger.Error("Failed to unlock consumer group after starting", zap.Error(err))
 		return err
 	}
+	m.notify(ManagerEvent{Event: GroupStarted, GroupId: groupId})
 	return nil
 }
 
diff --git a/pkg/common/consumer/managed_group.go b/pkg/common/consumer/managed_group.go
index 9408553d9..b71dc1a79 100644
--- a/pkg/common/consumer/managed_group.go
+++ b/pkg/common/consumer/managed_group.go
@@ -43,6 +43,7 @@ type managedGroup interface {
 	close() error
 	errors() chan error
 	processLock(*commands.CommandLock, bool) error
+	isStopped() bool
 }
 
 // managedGroupImpl implements the managedGroup interface
diff --git a/pkg/common/kafka/sarama/batch_get_offsets.go b/pkg/common/kafka/sarama/batch_get_offsets.go
new file mode 100644
index 000000000..bb0dc16ca
--- /dev/null
+++ b/pkg/common/kafka/sarama/batch_get_offsets.go
@@ -0,0 +1,83 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package sarama
+
+import (
+	"github.com/Shopify/sarama"
+)
+
+// GetOffsets queries the cluster to get the most recent available offset at the
+// given time (in milliseconds) for the given topics and partition
+// Time should be OffsetOldest for the earliest available offset,
+// OffsetNewest for the offset of the message that will be produced next, or a time.
+//
+// See sarama.Client.GetOffset for getting the offset of a single topic/partition combination
+func GetOffsets(client sarama.Client, topicPartitions map[string][]int32, time int64) (map[string]map[int32]int64, error) {
+	if client.Closed() {
+		return nil, sarama.ErrClosedClient
+	}
+
+	version := int16(0)
+	if client.Config().Version.IsAtLeast(sarama.V0_10_1_0) {
+		version = 1
+	}
+
+	offsets := make(map[string]map[int32]int64)
+
+	// Step 1: build one OffsetRequest instance per broker.
+	requests := make(map[*sarama.Broker]*sarama.OffsetRequest)
+
+	for topic, partitions := range topicPartitions {
+		offsets[topic] = make(map[int32]int64)
+
+		for _, partitionID := range partitions {
+			broker, err := client.Leader(topic, partitionID)
+			if err != nil {
+				return nil, err
+			}
+
+			request, ok := requests[broker]
+			if !ok {
+				request = &sarama.OffsetRequest{Version: version}
+				requests[broker] = request
+			}
+
+			request.AddBlock(topic, partitionID, time, 1)
+		}
+	}
+
+	// Step 2: send requests, one per broker, and collect offsets
+	for broker, request := range requests {
+		response, err := broker.GetAvailableOffsets(request)
+
+		if err != nil {
+			return nil, err
+		}
+
+		for topic, blocks := range response.Blocks {
+			for partitionID, block := range blocks {
+				if block.Err != sarama.ErrNoError {
+					return nil, block.Err
+				}
+
+				offsets[topic][partitionID] = block.Offset
+			}
+		}
+	}
+
+	return offsets, nil
+}
diff --git a/pkg/source/client/offsets.go b/pkg/source/client/offsets.go
index 0f8a1d2fe..dac1574a0 100644
--- a/pkg/source/client/offsets.go
+++ b/pkg/source/client/offsets.go
@@ -23,6 +23,8 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	"knative.dev/pkg/logging"
+
+	knsarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
 )
 
 // We want to make sure that ALL consumer group offsets are set before marking
@@ -54,6 +56,12 @@ func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string
 		topicPartitions[topic] = partitions
 	}
 
+	// Fetch topic offsets
+	topicOffsets, err := knsarama.GetOffsets(kafkaClient, topicPartitions, sarama.OffsetNewest)
+	if err != nil {
+		return fmt.Errorf("failed to get the topic offsets: %w", err)
+	}
+
 	// Look for uninitialized offset (-1)
 	offsets, err := kafkaAdminClient.ListConsumerGroupOffsets(consumerGroup, topicPartitions)
 	if err != nil {
@@ -62,19 +70,25 @@ func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string
 
 	dirty := false
 	for topic, partitions := range offsets.Blocks {
-		for partition, block := range partitions {
+		for partitionID, block := range partitions {
 			if block.Offset == -1 { // not initialized?
-				// Fetch the newest offset in the topic/partition and set it in the consumer group
-				offset, err := kafkaClient.GetOffset(topic, partition, sarama.OffsetNewest)
-				if err != nil {
-					return fmt.Errorf("failed to get the offset for topic %s and partition %d: %w", topic, partition, err)
+				partitionsOffsets, ok := topicOffsets[topic]
+				if !ok {
+					// topic may have been deleted. ignore.
+					continue
+				}
+
+				offset, ok := partitionsOffsets[partitionID]
+				if !ok {
+					// partition may have been deleted. ignore.
+					continue
 				}
 
-				logging.FromContext(ctx).Infow("initializing offset", zap.String("topic", topic), zap.Int32("partition", partition), zap.Int64("offset", offset))
+				logging.FromContext(ctx).Infow("initializing offset", zap.String("topic", topic), zap.Int32("partition", partitionID), zap.Int64("offset", offset))
 
-				pm, err := offsetManager.ManagePartition(topic, partition)
+				pm, err := offsetManager.ManagePartition(topic, partitionID)
 				if err != nil {
-					return fmt.Errorf("failed to create the partition manager for topic %s and partition %d: %w", topic, partition, err)
+					return fmt.Errorf("failed to create the partition manager for topic %s and partition %d: %w", topic, partitionID, err)
 				}
 
 				pm.MarkOffset(offset, "")


Test patch:
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
index e2adbbe17..8efe528ec 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
@@ -18,31 +18,36 @@ package controller
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"testing"
 	"time"
 
-	commonenv "knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
-
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
 	clientgotesting "k8s.io/client-go/testing"
 	"k8s.io/client-go/tools/record"
-	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/dispatcher"
-	reconciletesting "knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/testing"
-	"knative.dev/eventing-kafka/pkg/client/clientset/versioned"
-	fakeclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake"
-	"knative.dev/eventing-kafka/pkg/client/informers/externalversions"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/controller"
 	kncontroller "knative.dev/pkg/controller"
 	logtesting "knative.dev/pkg/logging/testing"
 	. "knative.dev/pkg/reconciler/testing"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	commonenv "knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/constants"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/dispatcher"
+	reconciletesting "knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/testing"
+	"knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	fakeclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake"
+	"knative.dev/eventing-kafka/pkg/client/informers/externalversions"
+	"knative.dev/eventing-kafka/pkg/common/consumer"
 )
 
 const (
@@ -58,25 +63,92 @@ func init() {
 // Test The NewController() Functionality
 func TestNewController(t *testing.T) {
 
-	// Test Data
+	for _, testCase := range []struct {
+		name          string
+		managerEvents chan consumer.ManagerEvent
+	}{
+		{
+			name:          "With manager events channel",
+			managerEvents: make(chan consumer.ManagerEvent),
+		},
+		{
+			name: "Nil manager events channel",
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			// Test Data
+			logger := logtesting.TestLogger(t).Desugar()
+			channelKey := "TestChannelKey"
+			mockDispatcher := &MockDispatcher{}
+			fakeKafkaChannelClientSet := fakeclientset.NewSimpleClientset()
+			fakeK8sClientSet := fake.NewSimpleClientset()
+			populateEnvironmentVariables(t)
+			kafkaInformerFactory := externalversions.NewSharedInformerFactory(fakeKafkaChannelClientSet, kncontroller.DefaultResyncPeriod)
+			kafkaChannelInformer := kafkaInformerFactory.Messaging().V1beta1().KafkaChannels()
+			stopChan := make(chan struct{})
+
+			// Perform The Test
+			c := NewController(logger, channelKey, mockDispatcher, kafkaChannelInformer, fakeK8sClientSet, fakeKafkaChannelClientSet, stopChan, testCase.managerEvents)
+
+			// Verify Results
+			assert.NotNil(t, c)
+
+			// Close The Channels
+			close(stopChan)
+			if testCase.managerEvents != nil {
+				close(testCase.managerEvents)
+			}
+
+			// Let the channel loops finish
+			time.Sleep(50 * time.Millisecond)
+
+			mockDispatcher.AssertExpectations(t)
+		})
+	}
+}
+
+// Test the processManagerEvents function
+func TestProcessManagerEvents(t *testing.T) {
+	mockDispatcher := &MockDispatcher{}
+	mockReconciler := &MockReconciler{}
 	logger := logtesting.TestLogger(t).Desugar()
-	channelKey := "TestChannelKey"
-	mockDispatcher := NewMockDispatcher(t)
 	fakeKafkaChannelClientSet := fakeclientset.NewSimpleClientset()
-	fakeK8sClientSet := fake.NewSimpleClientset()
 	populateEnvironmentVariables(t)
 	kafkaInformerFactory := externalversions.NewSharedInformerFactory(fakeKafkaChannelClientSet, kncontroller.DefaultResyncPeriod)
 	kafkaChannelInformer := kafkaInformerFactory.Messaging().V1beta1().KafkaChannels()
-	stopChan := make(chan struct{})
+	reconciler := &Reconciler{
+		logger:               logger,
+		dispatcher:           mockDispatcher,
+		kafkachannelInformer: kafkaChannelInformer.Informer(),
+		kafkachannelLister:   kafkaChannelInformer.Lister(),
+		kafkaClientSet:       fakeKafkaChannelClientSet,
+		impl:                 controller.NewImplFull(mockReconciler, controller.ControllerOptions{Logger: logger.Sugar()}),
+	}
+	events := make(chan consumer.ManagerEvent)
+
+	assert.NotNil(t, reconciler.processManagerEvents(nil))
+	reconciler.channelKey = "invalid/channel/key"
+	assert.NotNil(t, reconciler.processManagerEvents(events))
+	reconciler.channelKey = "test-namespace/test-name"
+	assert.Nil(t, reconciler.processManagerEvents(events))
+
+	// Send all of the supported event types to the events channel
+	events <- consumer.ManagerEvent{Event: consumer.GroupCreated, GroupId: "test-group-id"}
+	events <- consumer.ManagerEvent{Event: consumer.GroupStopped, GroupId: "test-group-id"}
+	events <- consumer.ManagerEvent{Event: consumer.GroupStarted, GroupId: "test-group-id"}
+	events <- consumer.ManagerEvent{Event: consumer.GroupClosed, GroupId: "test-group-id"}
 
-	// Perform The Test
-	c := NewController(logger, channelKey, mockDispatcher, kafkaChannelInformer, fakeK8sClientSet, fakeKafkaChannelClientSet, stopChan)
+	// Send an unexpected event type to the events channel
+	events <- consumer.ManagerEvent{Event: consumer.EventIndex(9999), GroupId: "test-group-id"}
 
-	// Verify Results
-	assert.NotNil(t, c)
+	// Close The Channels
+	close(events)
 
-	// Close The
-	close(stopChan)
+	// Let the channel loops finish
+	time.Sleep(50 * time.Millisecond)
+
+	mockDispatcher.AssertExpectations(t)
+	mockReconciler.AssertExpectations(t)
 }
 
 // Test KafkaChannel Controller Reconciliation
@@ -133,6 +205,21 @@ func TestAllCases(t *testing.T) {
 			},
 		},
 		{
+			Name: "channel ready, no change",
+			Objects: []runtime.Object{
+				reconciletesting.NewKafkaChannel(kcName, testNS,
+					reconciletesting.WithInitKafkaChannelConditions,
+					reconciletesting.WithKafkaChannelAddress("http://foobar"),
+					reconciletesting.WithKafkaChannelReady,
+					reconciletesting.WithSubscriber("1", "http://foobar"),
+					reconciletesting.WithSubscriberReady("1")),
+			},
+			Key:     kcKey,
+			WantErr: false,
+			WantEvents: []string{
+				Eventf(corev1.EventTypeNormal, channelReconciled, "KafkaChannel Reconciled"),
+			},
+		}, {
 			Name: "channel ready, 1 subscriber ready, add a 2nd one",
 			Objects: []runtime.Object{
 				reconciletesting.NewKafkaChannel(kcName, testNS,
@@ -160,15 +247,83 @@ func TestAllCases(t *testing.T) {
 				Eventf(corev1.EventTypeNormal, channelReconciled, "KafkaChannel Reconciled"),
 			},
 		},
+		{
+			Name: "channel ready, 1 subscriber ready, stopped, add 2nd one",
+			Objects: []runtime.Object{
+				reconciletesting.NewKafkaChannel(kcName, testNS,
+					reconciletesting.WithInitKafkaChannelConditions,
+					reconciletesting.WithKafkaChannelAddress("http://channel"),
+					reconciletesting.WithKafkaChannelReady,
+					reconciletesting.WithSubscriber("1", "http://foobar"),
+					reconciletesting.WithSubscriber("2", "http://foobar2"),
+					reconciletesting.WithSubscriberReady("1")),
+			},
+			Key:     kcKey,
+			WantErr: false,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+				Object: reconciletesting.NewKafkaChannel(kcName, testNS,
+					reconciletesting.WithInitKafkaChannelConditions,
+					reconciletesting.WithKafkaChannelReady,
+					reconciletesting.WithKafkaChannelAddress("http://channel"),
+					reconciletesting.WithSubscriber("1", "http://foobar"),
+					reconciletesting.WithSubscriber("2", "http://foobar2"),
+					reconciletesting.WithSubscriberNotReady("1", constants.GroupStoppedMessage),
+					reconciletesting.WithSubscriberReady("2"),
+				),
+			}},
+			WantEvents: []string{
+				Eventf(corev1.EventTypeNormal, channelReconciled, "KafkaChannel Reconciled"),
+			},
+			OtherTestData: map[string]interface{}{
+				"status": consumer.SubscriberStatusMap{types.UID("1"): consumer.SubscriberStatus{Stopped: true}},
+			},
+		},
+		{
+			Name: "channel ready, 1 subscriber ready, failed, add 2nd one",
+			Objects: []runtime.Object{
+				reconciletesting.NewKafkaChannel(kcName, testNS,
+					reconciletesting.WithInitKafkaChannelConditions,
+					reconciletesting.WithKafkaChannelAddress("http://channel"),
+					reconciletesting.WithKafkaChannelReady,
+					reconciletesting.WithSubscriber("1", "http://foobar"),
+					reconciletesting.WithSubscriber("2", "http://foobar2"),
+					reconciletesting.WithSubscriberReady("1")),
+			},
+			Key:     kcKey,
+			WantErr: false,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+				Object: reconciletesting.NewKafkaChannel(kcName, testNS,
+					reconciletesting.WithInitKafkaChannelConditions,
+					reconciletesting.WithKafkaChannelReady,
+					reconciletesting.WithKafkaChannelAddress("http://channel"),
+					reconciletesting.WithSubscriber("1", "http://foobar"),
+					reconciletesting.WithSubscriber("2", "http://foobar2"),
+					reconciletesting.WithSubscriberNotReady("1", "test error"),
+					reconciletesting.WithSubscriberReady("2"),
+				),
+			}},
+			WantEvents: []string{
+				Eventf(corev1.EventTypeWarning, channelReconcileFailed, "KafkaChannel Reconciliation Failed: some kafka subscribers failed to subscribe"),
+			},
+			OtherTestData: map[string]interface{}{
+				"status": consumer.SubscriberStatusMap{types.UID("1"): consumer.SubscriberStatus{Error: fmt.Errorf("test error")}},
+			},
+		},
 	}
 
-	table.Test(t, reconciletesting.MakeFactory(func(listers *reconciletesting.Listers, kafkaClient versioned.Interface, eventRecorder record.EventRecorder) controller.Reconciler {
+	table.Test(t, reconciletesting.MakeFactory(func(listers *reconciletesting.Listers,
+		kafkaClient versioned.Interface,
+		eventRecorder record.EventRecorder,
+		status consumer.SubscriberStatusMap,
+	) controller.Reconciler {
+		mockDispatcher := &MockDispatcher{}
+		mockDispatcher.On("UpdateSubscriptions", mock.Anything).Return(status)
 		return &Reconciler{
 			logger:               logtesting.TestLogger(t).Desugar(),
 			channelKey:           kcKey,
 			kafkachannelInformer: nil,
 			kafkachannelLister:   listers.GetKafkaChannelLister(),
-			dispatcher:           NewMockDispatcher(t),
+			dispatcher:           mockDispatcher,
 			recorder:             eventRecorder,
 			kafkaClientSet:       kafkaClient,
 		}
@@ -203,19 +358,34 @@ var _ dispatcher.Dispatcher = &MockDispatcher{}
 
 // Define The Mock Dispatcher
 type MockDispatcher struct {
-	t *testing.T
+	mock.Mock
 }
 
-// Mock Dispatcher Constructor
-func NewMockDispatcher(t *testing.T) MockDispatcher {
-	return MockDispatcher{t: t}
+func (m *MockDispatcher) Shutdown() {
+	m.Called()
 }
 
-func (m MockDispatcher) Shutdown() {
+func (m *MockDispatcher) UpdateSubscriptions(subscriberSpecs []eventingduck.SubscriberSpec) consumer.SubscriberStatusMap {
+	args := m.Called(subscriberSpecs)
+	return args.Get(0).(consumer.SubscriberStatusMap)
 }
 
-func (m MockDispatcher) UpdateSubscriptions(_ []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error {
-	return nil
+func (m *MockDispatcher) SecretChanged(ctx context.Context, secret *corev1.Secret) {
+	m.Called(ctx, secret)
 }
 
-func (m MockDispatcher) SecretChanged(_ context.Context, _ *corev1.Secret) {}
+//
+// Mock Reconciler Implementation
+//
+
+// Verify The Mock Reconciler Implements The Interface
+var _ controller.Reconciler = &MockReconciler{}
+
+// Define The Mock Dispatcher
+type MockReconciler struct {
+	mock.Mock
+}
+
+func (m *MockReconciler) Reconcile(ctx context.Context, key string) error {
+	return m.Called(ctx, key).Error(0)
+}
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
index 164c2b7a4..467e4126d 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
@@ -119,6 +119,7 @@ func TestShutdown(t *testing.T) {
 	// Note that consumerGroup2 does not have Close called because we are returning false for IsManaged()
 	consumerGroup3.On("Close").Return(nil)
 
+	mockManager.On("ClearNotifications").Return()
 	mockManager.On("IsManaged", groupId1).Return(true)
 	mockManager.On("IsManaged", groupId2).Return(false)
 	mockManager.On("IsManaged", groupId3).Return(true)
@@ -133,7 +134,6 @@ func TestShutdown(t *testing.T) {
 	consumerGroup1.AssertExpectations(t)
 	consumerGroup2.AssertExpectations(t)
 	consumerGroup3.AssertExpectations(t)
-	mockManager.AssertExpectations(t)
 
 	assert.Len(t, dispatcher.subscribers, 1) // One was unmanaged and should be ignored
 
@@ -143,6 +143,7 @@ func TestShutdown(t *testing.T) {
 	dispatcher.MetricsStoppedChan = make(chan struct{})
 	close(dispatcher.MetricsStoppedChan)
 	dispatcher.Shutdown()
+	mockManager.AssertExpectations(t)
 }
 
 // Test The UpdateSubscriptions() Functionality
@@ -179,16 +180,18 @@ func TestUpdateSubscriptions(t *testing.T) {
 
 	// Define The TestCase Struct
 	type TestCase struct {
-		name          string
-		fields        fields
-		createErr     error
-		consumeErr    bool
-		args          args
-		wantErrors    int
-		wantFailure   bool
-		expectStarted []string
-		expectErrors  []string
-		expectManaged []string
+		name            string
+		fields          fields
+		createErr       error
+		consumeErr      bool
+		args            args
+		wantErrors      int
+		wantFailure     bool
+		wantStop        bool
+		expectStarted   []string
+		expectErrors    []string
+		expectIsManaged []string
+		expectIsStopped []string
 	}
 
 	// Create The Test Cases
@@ -204,9 +207,9 @@ func TestUpdateSubscriptions(t *testing.T) {
 					{UID: uid123},
 				},
 			},
-			expectStarted: []string{id123},
-			expectErrors:  []string{id123},
-			expectManaged: []string{id123},
+			expectStarted:   []string{id123},
+			expectErrors:    []string{id123},
+			expectIsManaged: []string{id123},
 		},
 		{
 			name: "Create ConsumerGroup Error",
@@ -234,10 +237,10 @@ func TestUpdateSubscriptions(t *testing.T) {
 					{UID: uid123},
 				},
 			},
-			consumeErr:    true,
-			expectStarted: []string{id123},
-			expectErrors:  []string{id123},
-			expectManaged: []string{id123},
+			consumeErr:      true,
+			expectStarted:   []string{id123},
+			expectErrors:    []string{id123},
+			expectIsManaged: []string{id123},
 		},
 		{
 			name: "Invalid Config",
@@ -266,9 +269,10 @@ func TestUpdateSubscriptions(t *testing.T) {
 					{UID: uid456},
 				},
 			},
-			expectStarted: []string{id456},
-			expectErrors:  []string{id456},
-			expectManaged: []string{id123, id456},
+			expectStarted:   []string{id456},
+			expectErrors:    []string{id456},
+			expectIsManaged: []string{id123, id456},
+			expectIsStopped: []string{id123},
 		},
 		{
 			name: "Add And Remove Subscriptions",
@@ -285,9 +289,27 @@ func TestUpdateSubscriptions(t *testing.T) {
 					{UID: uid789},
 				},
 			},
-			expectStarted: []string{id789},
-			expectErrors:  []string{id789},
-			expectManaged: []string{id123, id456, id789},
+			expectStarted:   []string{id789},
+			expectErrors:    []string{id789},
+			expectIsManaged: []string{id123, id456, id789},
+			expectIsStopped: []string{id456},
+		},
+		{
+			name: "No Add Or Remove, Group Stopped",
+			fields: fields{
+				DispatcherConfig: dispatcherConfig,
+				subscribers: map[types.UID]*SubscriberWrapper{
+					uid123: createSubscriberWrapper(uid123),
+				},
+			},
+			args: args{
+				subscriberSpecs: []eventingduck.SubscriberSpec{
+					{UID: uid123},
+				},
+			},
+			expectIsManaged: []string{id123},
+			expectIsStopped: []string{id123},
+			wantStop:        true,
 		},
 		{
 			name: "Remove Penultimate Subscription",
@@ -303,7 +325,8 @@ func TestUpdateSubscriptions(t *testing.T) {
 					{UID: uid123},
 				},
 			},
-			expectManaged: []string{id123, id456},
+			expectIsManaged: []string{id123, id456},
+			expectIsStopped: []string{id123},
 		},
 		{
 			name: "Remove Last Subscription",
@@ -316,7 +339,7 @@ func TestUpdateSubscriptions(t *testing.T) {
 			args: args{
 				subscriberSpecs: []eventingduck.SubscriberSpec{},
 			},
-			expectManaged: []string{id123},
+			expectIsManaged: []string{id123},
 		},
 	}
 
@@ -337,15 +360,21 @@ func TestUpdateSubscriptions(t *testing.T) {
 				errorSource <- fmt.Errorf("consume error")
 			}
 
-			for _, id := range testCase.expectStarted {
-				mockManager.On("StartConsumerGroup", "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))
-			}
-			for _, id := range testCase.expectManaged {
-				mockManager.On("IsManaged", "kafka."+id).Return(true)
-				mockManager.On("CloseConsumerGroup", "kafka."+id).Return(nil)
+			if !testCase.wantFailure {
+				mockManager.On("ClearNotifications").Return()
+				for _, id := range testCase.expectStarted {
+					mockManager.On("StartConsumerGroup", "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))
+				}
+				for _, id := range testCase.expectIsManaged {
+					mockManager.On("IsManaged", "kafka."+id).Return(true)
+					mockManager.On("CloseConsumerGroup", "kafka."+id).Return(nil)
+				}
+				for _, id := range testCase.expectIsStopped {
+					mockManager.On("IsStopped", "kafka."+id).Return(testCase.wantStop)
+				}
 			}
 
 			// Perform The Test
@@ -357,7 +386,16 @@ func TestUpdateSubscriptions(t *testing.T) {
 				assert.Nil(t, result)
 			} else {
 				// Verify Results
-				assert.Equal(t, testCase.wantErrors, len(result))
+				stoppedCount := 0
+				for _, status := range result {
+					if status.Stopped {
+						stoppedCount++
+					}
+				}
+				assert.Equal(t, testCase.wantErrors, result.FailedCount())
+				if testCase.wantStop {
+					assert.Equal(t, len(testCase.expectIsStopped), stoppedCount)
+				}
 
 				// Verify The Dispatcher's Tracking Of Subscribers Matches Specified State
 				assert.Equal(t, len(testCase.args.subscriberSpecs)-testCase.wantErrors, len(dispatcher.subscribers))
@@ -530,7 +568,8 @@ 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, events := NewDispatcher(dispatcherConfig, serverHandler)
+	assert.NotNil(t, events)
 	serverHandler.AssertExpectations(t)
 
 	// Verify State
diff --git a/pkg/channel/distributed/dispatcher/testing/factory.go b/pkg/channel/distributed/dispatcher/testing/factory.go
index 992fb54a5..ef5c72434 100644
--- a/pkg/channel/distributed/dispatcher/testing/factory.go
+++ b/pkg/channel/distributed/dispatcher/testing/factory.go
@@ -22,9 +22,11 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/tools/record"
 	"knative.dev/eventing-kafka/pkg/client/clientset/versioned"
-	fakeclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake"
+	"knative.dev/eventing-kafka/pkg/common/consumer"
 	"knative.dev/pkg/controller"
 	. "knative.dev/pkg/reconciler/testing"
+
+	fakeclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake"
 )
 
 const (
@@ -34,7 +36,12 @@ const (
 )
 
 // Ctor functions create a k8s controller with given params.
-type Ctor func(listers *Listers, kafkaClient versioned.Interface, eventRecorder record.EventRecorder) controller.Reconciler
+type Ctor func(
+	listers *Listers,
+	kafkaClient versioned.Interface,
+	eventRecorder record.EventRecorder,
+	status consumer.SubscriberStatusMap,
+) controller.Reconciler
 
 // MakeFactory creates a reconciler factory with fake clients and controller created by `ctor`.
 func MakeFactory(ctor Ctor) Factory {
@@ -48,10 +55,17 @@ func MakeFactory(ctor Ctor) Factory {
 			_ = addTo(dynamicScheme)
 		}
 
+		var status consumer.SubscriberStatusMap
+
+		statusInt, ok := r.OtherTestData["status"]
+		if ok {
+			status = statusInt.(consumer.SubscriberStatusMap)
+		}
+
 		eventRecorder := record.NewFakeRecorder(maxEventBufferSize)
 
 		// Set up our Controller from the fakes.
-		c := ctor(&ls, client, eventRecorder)
+		c := ctor(&ls, client, eventRecorder, status)
 
 		actionRecorderList := ActionRecorderList{client}
 		eventList := EventList{Recorder: eventRecorder}
diff --git a/pkg/channel/distributed/dispatcher/testing/kafkachannel.go b/pkg/channel/distributed/dispatcher/testing/kafkachannel.go
index 01198af63..5934365f6 100644
--- a/pkg/channel/distributed/dispatcher/testing/kafkachannel.go
+++ b/pkg/channel/distributed/dispatcher/testing/kafkachannel.go
@@ -99,3 +99,16 @@ func WithSubscriberReady(uid types.UID) KafkaChannelOption {
 		})
 	}
 }
+
+func WithSubscriberNotReady(uid types.UID, message string) KafkaChannelOption {
+	return func(kafkachannel *v1beta1.KafkaChannel) {
+		if kafkachannel.Status.SubscribableStatus.Subscribers == nil {
+			kafkachannel.Status.SubscribableStatus.Subscribers = []eventingduck.SubscriberStatus{}
+		}
+		kafkachannel.Status.SubscribableStatus.Subscribers = append(kafkachannel.Status.SubscribableStatus.Subscribers, eventingduck.SubscriberStatus{
+			Ready:   corev1.ConditionFalse,
+			UID:     uid,
+			Message: message,
+		})
+	}
+}
diff --git a/pkg/common/consumer/consumer_manager_test.go b/pkg/common/consumer/consumer_manager_test.go
index 1acf245c9..86b75fdbc 100644
--- a/pkg/common/consumer/consumer_manager_test.go
+++ b/pkg/common/consumer/consumer_manager_test.go
@@ -19,6 +19,8 @@ package consumer
 import (
 	"context"
 	"fmt"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -244,9 +246,11 @@ func TestErrors(t *testing.T) {
 		t.Run(testCase.name, func(t *testing.T) {
 			manager, _, _, server := getManagerWithMockGroup(t, testCase.groupId, false)
 			valid := manager.IsManaged(testCase.groupId)
+			stopped := manager.IsStopped(testCase.groupId)
 			mgrErrors := manager.Errors(testCase.groupId)
 			assert.Equal(t, testCase.expectErr, mgrErrors == nil)
 			assert.Equal(t, !testCase.expectErr, valid)
+			assert.False(t, stopped) // Not actually using a stopped group in this test
 			server.AssertExpectations(t)
 		})
 	}
@@ -452,6 +456,97 @@ func TestNotifications(t *testing.T) {
 	}
 }
 
+func TestManagerEvents(t *testing.T) {
+	defer restoreNewConsumerGroup(newConsumerGroup) // must use if calling getManagerWithMockGroup in the test
+	for _, testCase := range []struct {
+		name        string
+		channels    int
+		listen      bool
+		expectEvent int32
+		expectClose int32
+	}{
+		{
+			name: "Notify Zero Channels",
+		},
+		{
+			name:        "Notify One Channel",
+			channels:    1,
+			listen:      true,
+			expectEvent: 1,
+			expectClose: 1,
+		},
+		{
+			name:        "Notify Two Channels",
+			channels:    2,
+			listen:      true,
+			expectEvent: 2,
+			expectClose: 2,
+		},
+		{
+			name:        "Notify Ten Channels",
+			channels:    10,
+			listen:      true,
+			expectEvent: 10,
+			expectClose: 10,
+		},
+		{
+			name:        "Notify Non-Listening Channel",
+			channels:    1,
+			expectClose: 1,
+		},
+		{
+			name:        "Notify Five Non-Listening Channels",
+			channels:    5,
+			expectClose: 5,
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			manager, _, _, _ := getManagerWithMockGroup(t, "", false)
+			impl := manager.(*kafkaConsumerGroupManagerImpl)
+
+			var notifyChannels []<-chan ManagerEvent
+			for i := 0; i < testCase.channels; i++ {
+				notifyChannels = append(notifyChannels, manager.GetNotificationChannel())
+			}
+
+			waitGroup := sync.WaitGroup{}
+			count := int32(0)
+
+			if testCase.listen {
+				waitGroup.Add(len(notifyChannels))
+				for _, notifyChan := range notifyChannels {
+					go func(channel <-chan ManagerEvent) {
+						<-channel
+						atomic.AddInt32(&count, 1)
+						waitGroup.Done()
+					}(notifyChan)
+				}
+			}
+
+			time.Sleep(50 * time.Millisecond)
+			impl.notify(ManagerEvent{Event: GroupCreated, GroupId: "created-group-id"})
+			waitGroup.Wait()
+			assert.Equal(t, testCase.expectEvent, count)
+
+			waitGroup = sync.WaitGroup{}
+			count = int32(0)
+			waitGroup.Add(len(notifyChannels))
+			for _, notifyChan := range notifyChannels {
+				go func(channel <-chan ManagerEvent) {
+					<-channel
+					atomic.AddInt32(&count, 1)
+					waitGroup.Done()
+				}(notifyChan)
+			}
+
+			time.Sleep(50 * time.Millisecond)
+			manager.ClearNotifications()
+			waitGroup.Wait()
+			assert.Equal(t, testCase.expectClose, count)
+		})
+	}
+}
+
 func TestProcessAsyncGroupNotification_BadMessage(t *testing.T) {
 	cmdFunctionCalled := false
 	cmdFunction := func(lock *commands.CommandLock, groupId string) error {
diff --git a/pkg/common/consumer/managed_group_test.go b/pkg/common/consumer/managed_group_test.go
index c01312d88..6d684065c 100644
--- a/pkg/common/consumer/managed_group_test.go
+++ b/pkg/common/consumer/managed_group_test.go
@@ -32,6 +32,8 @@ import (
 	kafkatesting "knative.dev/eventing-kafka/pkg/common/kafka/testing"
 )
 
+const shortTimeout = 200 * time.Millisecond
+
 func TestManagedGroup(t *testing.T) {
 	for _, testCase := range []struct {
 		name     string
@@ -91,7 +93,6 @@ func TestManagedGroup(t *testing.T) {
 func TestProcessLock(t *testing.T) {
 	defer restoreNewConsumerGroup(newConsumerGroup) // must use if calling getManagerWithMockGroup in the test
 	const newToken = "new-token"
-	const shortTimeout = 60 * time.Millisecond
 
 	for _, testCase := range []struct {
 		name            string
@@ -193,8 +194,6 @@ func TestProcessLock(t *testing.T) {
 }
 
 func TestResetLockTimer(t *testing.T) {
-	const shortTimeout = 60 * time.Millisecond
-
 	for _, testCase := range []struct {
 		name          string
 		cancelTimer   func()
@@ -445,8 +444,6 @@ func TestClose(t *testing.T) {
 }
 
 func TestTransferErrors(t *testing.T) {
-	const shortTimeout = 60 * time.Millisecond
-
 	for _, testCase := range []struct {
 		name       string
 		stopGroup  bool
@@ -555,3 +552,7 @@ func (m *mockManagedGroup) errors() chan error {
 func (m *mockManagedGroup) processLock(cmdLock *commands.CommandLock, lock bool) error {
 	return m.Called(cmdLock, lock).Error(0)
 }
+
+func (m *mockManagedGroup) isStopped() bool {
+	return m.Called().Bool(0)
+}
diff --git a/pkg/common/consumer/testing/mocks.go b/pkg/common/consumer/testing/mocks.go
index 86a521225..73e6bd29a 100644
--- a/pkg/common/consumer/testing/mocks.go
+++ b/pkg/common/consumer/testing/mocks.go
@@ -57,31 +57,38 @@ func NewMockConsumerGroupManager() *MockConsumerGroupManager {
 var _ consumer.KafkaConsumerGroupManager = (*MockConsumerGroupManager)(nil)
 
 func (m *MockConsumerGroupManager) Reconfigure(brokers []string, config *sarama.Config) error {
-	args := m.Called(brokers, config)
-	return args.Error(0)
+	return m.Called(brokers, config).Error(0)
 }
 
 func (m *MockConsumerGroupManager) StartConsumerGroup(groupId string, topics []string, logger *zap.SugaredLogger,
 	handler consumer.KafkaConsumerHandler, options ...consumer.SaramaConsumerHandlerOption) error {
-	args := m.Called(groupId, topics, logger, handler, options)
-	return args.Error(0)
+	return m.Called(groupId, topics, logger, handler, options).Error(0)
 }
 
 func (m *MockConsumerGroupManager) CloseConsumerGroup(groupId string) error {
-	args := m.Called(groupId)
 	if group, ok := m.Groups[groupId]; ok {
 		_ = group.Close()
 		delete(m.Groups, groupId)
 	}
-	return args.Error(0)
+	return m.Called(groupId).Error(0)
 }
 
 func (m *MockConsumerGroupManager) IsManaged(groupId string) bool {
-	args := m.Called(groupId)
-	return args.Bool(0)
+	return m.Called(groupId).Bool(0)
+}
+
+func (m *MockConsumerGroupManager) IsStopped(groupId string) bool {
+	return m.Called(groupId).Bool(0)
 }
 
 func (m *MockConsumerGroupManager) Errors(groupId string) <-chan error {
-	args := m.Called(groupId)
-	return args.Get(0).(<-chan error)
+	return m.Called(groupId).Get(0).(<-chan error)
+}
+
+func (m *MockConsumerGroupManager) GetNotificationChannel() <-chan consumer.ManagerEvent {
+	return m.Called().Get(0).(<-chan consumer.ManagerEvent)
+}
+
+func (m *MockConsumerGroupManager) ClearNotifications() {
+	_ = m.Called()
 }
diff --git a/pkg/common/kafka/sarama/batch_get_offsets_test.go b/pkg/common/kafka/sarama/batch_get_offsets_test.go
new file mode 100644
index 000000000..5ec05148c
--- /dev/null
+++ b/pkg/common/kafka/sarama/batch_get_offsets_test.go
@@ -0,0 +1,118 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package sarama
+
+import (
+	"testing"
+	"time"
+
+	"github.com/Shopify/sarama"
+)
+
+func TestGetOffset(t *testing.T) {
+	testCases := map[string]struct {
+		topicPartitions map[string][]int32
+		topicOffsets    map[string]map[int32]int64
+	}{
+		"no topics": {
+			topicPartitions: map[string][]int32{},
+			topicOffsets:    map[string]map[int32]int64{},
+		},
+		"one topic, one partition": {
+			topicPartitions: map[string][]int32{"my-topic": {0}},
+			topicOffsets: map[string]map[int32]int64{
+				"my-topic": {
+					0: 5,
+				},
+			},
+		},
+		"several topics, several partitions": {
+			topicPartitions: map[string][]int32{
+				"my-topic":   {0, 1},
+				"my-topic-2": {0, 1, 2},
+				"my-topic-3": {0, 1, 2, 3}},
+			topicOffsets: map[string]map[int32]int64{
+				"my-topic":   {0: 5, 1: 7},
+				"my-topic-2": {0: 5, 1: 7, 2: 9},
+				"my-topic-3": {0: 5, 1: 7, 2: 2, 3: 10},
+			},
+		},
+	}
+
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			broker := sarama.NewMockBroker(t, 1)
+			defer broker.Close()
+			offsetResponse := sarama.NewMockOffsetResponse(t).SetVersion(1)
+			for topic, partitions := range tc.topicOffsets {
+				for partition, offset := range partitions {
+					offsetResponse = offsetResponse.SetOffset(topic, partition, -1, offset)
+				}
+			}
+
+			metadataResponse := sarama.NewMockMetadataResponse(t).
+				SetController(broker.BrokerID()).
+				SetBroker(broker.Addr(), broker.BrokerID())
+
+			for topic, partitions := range tc.topicOffsets {
+				for partition := range partitions {
+					metadataResponse = metadataResponse.SetLeader(topic, partition, broker.BrokerID())
+				}
+			}
+
+			broker.SetHandlerByMap(map[string]sarama.MockResponse{
+				"OffsetRequest":   offsetResponse,
+				"MetadataRequest": metadataResponse,
+			})
+
+			config := sarama.NewConfig()
+			config.Version = sarama.MaxVersion
+
+			sc, err := sarama.NewClient([]string{broker.Addr()}, config)
+			if err != nil {
+				t.Errorf("unexpected error: %v", err)
+			}
+			defer sc.Close()
+
+			offsets, err := GetOffsets(sc, tc.topicPartitions, sarama.OffsetNewest)
+
+			if err != nil {
+				t.Errorf("unexpected error: %v", err)
+			}
+
+			if len(tc.topicOffsets) != len(offsets) {
+				t.Errorf("unexpected number of topics. wanted %d, got %d", len(tc.topicOffsets), len(offsets))
+			}
+
+			for topic, topicOffsets := range offsets {
+				if len(tc.topicOffsets[topic]) != len(topicOffsets) {
+					t.Errorf("unexpected number of partitions. wanted %d, got %d", len(tc.topicOffsets[topic]), len(topicOffsets))
+				}
+
+				for partitionID, offset := range topicOffsets {
+					if tc.topicOffsets[topic][partitionID] != offset {
+						t.Errorf("unexpected offset. wanted %d, got %d", tc.topicOffsets[topic][partitionID], offset)
+					}
+				}
+			}
+
+		})
+	}
+
+	// Seems like there is no way to check the broker has been really closed. So wait few seconds
+	time.Sleep(2 * time.Second)
+}