Repository: knative-sandbox/eventing-kafka
Pull request #755 (2021-07-07 22:10:54)

Change kafkachannel status when paused
Subtask of #588 to update the "subscribers" section of the status block for KafkaChannel resources when a ConsumerGroup is started or stopped.

## Proposed Changes
🎁  Added Go channel notification option to the KafkaConsumerGroupManager (GetNotificationChannel) so that external entities may be informed of changes to a managed ConsumerGroup's status (create/start/stop/close)
🎁. Modified the distributed Dispatcher's KafkaChannel controller to use this new channel from the Manager and call EnqueueKey in order to trigger a reconciliation when Consumer Groups are started or stopped.
🎁  Modified the distributed Dispatcher's UpdateSubscriptions() call to return a map of status infomation that includes "stopped," so that the reconciler can use that information to update the subscribers status block properly.

**Notes**

- When the KafkaChannel's status is updated, the related Knative Eventing subscription resource is automatically (no change by this PR) changed to a status of "False" with a reason of SubscriptionNotMarkedReadyByChannel ("Subscription marked by Channel as False").  I believe this is sufficient.
- The ResetOffset Controller is still "inactive" and has not been added to any Channels/Sources/etc.


**Release Note**
Release Notes will be provided near the end of the ResetOffset feature development.

**Docs**
Documentation will be provided near the end of the ResetOffset feature development.

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/755#" title="Author self-approved">eric-sap</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)~~ [eric-sap]

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":[]} -->
eric-sap: /hold
holding for unit test race condition experiment
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/channel/distributed/dispatcher/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/755/pull-knative-sandbox-eventing-kafka-go-coverage/1412896943473954816/artifacts/line-cov.html#file0) | 71.8% | 86.8% | 15.0
[pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/755/pull-knative-sandbox-eventing-kafka-go-coverage/1412896943473954816/artifacts/line-cov.html#file1) | 97.5% | 97.6% | 0.1

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#755](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (90da27e) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/699e408976ce0a86bd7c4fbf6790941e2bd0dec8?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (699e408) will **increase** coverage by `1.94%`.
> The diff coverage is `95.67%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755/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/755?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     #755      +/-   ##
==========================================
+ Coverage   73.84%   75.78%   +1.94%     
==========================================
  Files         133      141       +8     
  Lines        5880     6405     +525     
==========================================
+ Hits         4342     4854     +512     
- Misses       1315     1336      +21     
+ Partials      223      215       -8     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [.../channel/distributed/controller/util/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci91dGlsL2Rpc3BhdGNoZXIuZ28=) | `100.00% <ø> (ø)` | |
| [pkg/common/commands/resetoffset/controller/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9jb250cm9sbGVyL3V0aWwuZ28=) | `66.66% <66.66%> (ø)` | |
| [...ommon/commands/resetoffset/controller/dataplane.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9jb250cm9sbGVyL2RhdGFwbGFuZS5nbw==) | `78.08% <78.08%> (ø)` | |
| [...mmon/commands/resetoffset/controller/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9jb250cm9sbGVyL2NvbnRyb2xsZXIuZ28=) | `81.81% <80.00%> (-3.90%)` | :arrow_down: |
| [...mmon/commands/resetoffset/controller/reconciler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9jb250cm9sbGVyL3JlY29uY2lsZXIuZ28=) | `96.47% <93.87%> (+1.47%)` | :arrow_up: |
| [pkg/common/controlprotocol/controlprotocol.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb250cm9scHJvdG9jb2wvY29udHJvbHByb3RvY29sLmdv) | `96.96% <96.96%> (ø)` | |
| [...on/commands/resetoffset/refmappers/subscription.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9yZWZtYXBwZXJzL3N1YnNjcmlwdGlvbi5nbw==) | `95.83% <97.87%> (+3.33%)` | :arrow_up: |
| [pkg/apis/kafka/v1alpha1/reset\_offset\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVzZXRfb2Zmc2V0X2xpZmVjeWNsZS5nbw==) | `100.00% <100.00%> (ø)` | |
| [...hannel/distributed/controller/util/subscription.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci91dGlsL3N1YnNjcmlwdGlvbi5nbw==) | `100.00% <100.00%> (ø)` | |
| [.../distributed/dispatcher/controller/kafkachannel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9jb250cm9sbGVyL2thZmthY2hhbm5lbC5nbw==) | `83.48% <100.00%> (+19.11%)` | :arrow_up: |
| ... and [18 more](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755/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/755?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/755?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [8d1daa0...90da27e](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/755?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).


Patch:
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/go.mod b/go.mod
index dd5eeee4c..64b2bdf7a 100644
--- a/go.mod
+++ b/go.mod
@@ -37,7 +37,7 @@ require (
 	knative.dev/control-protocol v0.0.0-20210701061703-8ddeca27693a
 	knative.dev/eventing v0.24.1-0.20210706174220-9ced57400dce
 	knative.dev/hack v0.0.0-20210622141627-e28525d8d260
-	knative.dev/networking v0.0.0-20210707064918-2a685f4fdd5a
+	knative.dev/networking v0.0.0-20210708015022-4e655b7fa1c3
 	knative.dev/pkg v0.0.0-20210706174620-fe90576475ca
-	knative.dev/reconciler-test v0.0.0-20210707095218-5879dd7b78c8
+	knative.dev/reconciler-test v0.0.0-20210707164418-32f0df0c7399
 )
diff --git a/go.sum b/go.sum
index ba249d137..241bd4ac2 100644
--- a/go.sum
+++ b/go.sum
@@ -1274,16 +1274,16 @@ knative.dev/eventing v0.24.1-0.20210706174220-9ced57400dce/go.mod h1:93VHah2Xghr
 knative.dev/hack v0.0.0-20210622141627-e28525d8d260 h1:f2eMtOubAOc/Q7JlvFPDKXiPlJVK+VpX2Cot8hRzCgQ=
 knative.dev/hack v0.0.0-20210622141627-e28525d8d260/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210622141627-e28525d8d260/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/networking v0.0.0-20210707064918-2a685f4fdd5a h1:AHYs6Yg6LrW8AaAbaSmfdXYX1so3lsE7bXTGo/K/RSk=
-knative.dev/networking v0.0.0-20210707064918-2a685f4fdd5a/go.mod h1:ljfndmG90nC/0NpzCMsrvru2uf1IUx4HFPLgRgqzMys=
+knative.dev/networking v0.0.0-20210708015022-4e655b7fa1c3 h1:8WM46YeGCGLP2k67+0CdsrJQUI+Qd1e81LSJwdc5HMU=
+knative.dev/networking v0.0.0-20210708015022-4e655b7fa1c3/go.mod h1:ljfndmG90nC/0NpzCMsrvru2uf1IUx4HFPLgRgqzMys=
 knative.dev/pkg v0.0.0-20210622173328-dd0db4b05c80/go.mod h1:kGegTnbZ+ljFjAE3E1+8wgaH2LMv8qYi+72o3F3cbdc=
 knative.dev/pkg v0.0.0-20210628225612-51cfaabbcdf6/go.mod h1:kGegTnbZ+ljFjAE3E1+8wgaH2LMv8qYi+72o3F3cbdc=
 knative.dev/pkg v0.0.0-20210701025203-30f9568e894e/go.mod h1:kGegTnbZ+ljFjAE3E1+8wgaH2LMv8qYi+72o3F3cbdc=
 knative.dev/pkg v0.0.0-20210706174620-fe90576475ca h1:WF0VUpn7S8RvhQ1Q419NAlI+iFJUBTDmVR7lbMzhGMk=
 knative.dev/pkg v0.0.0-20210706174620-fe90576475ca/go.mod h1:kGegTnbZ+ljFjAE3E1+8wgaH2LMv8qYi+72o3F3cbdc=
 knative.dev/reconciler-test v0.0.0-20210630182710-2a6d91dfee1e/go.mod h1:4wqv2WyWUC5yhTesRUVwgjv/fHTHny1RYBfdB6tVDok=
-knative.dev/reconciler-test v0.0.0-20210707095218-5879dd7b78c8 h1:fL/JzFNidKb9EWWrY8yfY0KVm2CGpLdL2SfrLife7Po=
-knative.dev/reconciler-test v0.0.0-20210707095218-5879dd7b78c8/go.mod h1:4wqv2WyWUC5yhTesRUVwgjv/fHTHny1RYBfdB6tVDok=
+knative.dev/reconciler-test v0.0.0-20210707164418-32f0df0c7399 h1:qPbQKtiWTL2Pw5r44CgxN9wxy7SN+drHlJMxvL7D6DA=
+knative.dev/reconciler-test v0.0.0-20210707164418-32f0df0c7399/go.mod h1:4wqv2WyWUC5yhTesRUVwgjv/fHTHny1RYBfdB6tVDok=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
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/vendor/modules.txt b/vendor/modules.txt
index 92d5b8784..379b53aad 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1241,7 +1241,7 @@ knative.dev/eventing/test/upgrade/prober/wathola/sender
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/networking v0.0.0-20210707064918-2a685f4fdd5a
+# knative.dev/networking v0.0.0-20210708015022-4e655b7fa1c3
 ## explicit
 knative.dev/networking/pkg/prober
 # knative.dev/pkg v0.0.0-20210706174620-fe90576475ca
@@ -1351,7 +1351,7 @@ knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/conversion
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20210707095218-5879dd7b78c8
+# knative.dev/reconciler-test v0.0.0-20210707164418-32f0df0c7399
 ## explicit
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


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/vendor/knative.dev/reconciler-test/pkg/eventshub/options.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/options.go
index 5b2c4a94d..f3f8f01ab 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/options.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/options.go
@@ -113,6 +113,13 @@ func DropFirstN(n uint) EventsHubOption {
 	)
 }
 
+// DropEventsResponseCode will cause the receiver to reply with the specific status code to the dropped events
+func DropEventsResponseCode(code int) EventsHubOption {
+	return compose(
+		envOption("SKIP_RESPONSE_CODE", strconv.Itoa(code)),
+	)
+}
+
 // --- Sender options
 
 // InitialSenderDelay defines how much the sender has to wait, when started, before start sending events.
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/prober.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/prober.go
index 4050e0d37..8f18f7662 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/prober.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/prober.go
@@ -103,6 +103,11 @@ func (p *EventProber) ReceiversRejectFirstN(n uint) {
 	p.receiverOptions = append(p.receiverOptions, DropFirstN(n))
 }
 
+// ReceiversRejectResponseCode adds DropEventsResponseCode to the default config for new receivers.
+func (p *EventProber) ReceiversRejectResponseCode(code int) {
+	p.receiverOptions = append(p.receiverOptions, DropEventsResponseCode(code))
+}
+
 // ReceiversHaveResponseDelay adds ResponseWaitTime to the default config for
 // new receivers.
 func (p *EventProber) ReceiversHaveResponseDelay(delay time.Duration) {
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/receiver/receiver.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/receiver/receiver.go
index daf4da0db..314244b70 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/receiver/receiver.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/receiver/receiver.go
@@ -47,6 +47,7 @@ type Receiver struct {
 	replyFunc        func(context.Context, http.ResponseWriter, eventshub.EventInfo)
 	counter          *dropevents.CounterHandler
 	responseWaitTime time.Duration
+	skipResponseCode int
 }
 
 type envConfig struct {
@@ -78,6 +79,9 @@ type envConfig struct {
 	// If events should be dropped according to Linear policy, this controls
 	// how many events are dropped.
 	SkipCounter uint64 `envconfig:"SKIP_COUNTER" default:"0" required:"false"`
+
+	// If events should be dropped, specify the HTTP response code here.
+	SkipResponseCode int `envconfig:"SKIP_RESPONSE_CODE" default:"409" required:"false"`
 }
 
 func NewFromEnv(ctx context.Context, eventLogs *eventshub.EventLogs) *Receiver {
@@ -121,6 +125,7 @@ func NewFromEnv(ctx context.Context, eventLogs *eventshub.EventLogs) *Receiver {
 		replyFunc:        replyFunc,
 		counter:          counter,
 		responseWaitTime: responseWaitTime,
+		skipResponseCode: env.SkipResponseCode,
 	}
 }
 
@@ -211,7 +216,7 @@ func (o *Receiver) ServeHTTP(writer http.ResponseWriter, request *http.Request)
 
 	if shouldSkip {
 		// Trigger a redelivery
-		writer.WriteHeader(http.StatusConflict)
+		writer.WriteHeader(o.skipResponseCode)
 	} else {
 		o.replyFunc(o.ctx, writer, eventInfo)
 	}