Repository: knative-sandbox/eventing-kafka
Pull request #561 (2021-04-21 16:03:09)

[distributed kafkachannel] Distributed Dispatcher with common/consumer
In order to better understand our options for attempting to further consolidate some of the ConsumerGroup logic between the _consolidated_ and _distributed_ Dispatcher implementations, I have created this POC to switch the _distributed_ implementation to use the **common/consumer** logic. 

## Proposed Changes

- _distributed_ Dispatcher uses the common/consumer KafkaConsumerGroupFactory instead of it's own Sarama ConsumerGroup lifecycle management.
- _distributed_ Dispatcher's Handler now implements the common/consumer KafkaConsumerHandler interface instead of the Sarama interface.
- Moved / enhanced some _consolidated_ mocks to common/consumer/testing package.
- Fixed the _distributed_ unit tests to align with new implementation

## TODOs

- Minor nit but the common/consumer KafkaConsumerHandler method `GetConsumerGroup()` function should be renamed `GetConsumerGroupId()` to better express what it does.  Also, that function exists only for a single log message in the factory so not sure how useful it is.
- The common/consumer logic (e.g. `SetReady()`) expects to be used with the current "Prober" implementation for tracking ConsumerGroup status.  The _distributed_ Dispatcher does not make use of this so there are some loose ends.  Not sure it would be worthwhile to implement support for the "Prober" since it is expected to be replaced with a new control-protocol based implementation in the near future?
- Importantly, the error returned from the `Handle()` function is used in the Prober implementation to set ConsumerGroup status.  While this isn't really an issue in this POC it is something we need to sort out if we move forward with this idea.  The _distributed_ implementation WILL mark the offset of failed messages in order to move on and not block the queue and we wouldn't want an error in sending an event to result in marking the ConsumerGroup status as failed/error.  We could always "swallow" the error inside the `Handle()` implementation but that doesn't seem correct.  Just a difference to iron out - shouldn't be a huge deal.
- I've only performed simple vanilla testing and would need to more closely confirm no message loss on shutdown/restarts etc.



knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **travis-minke-sap** after the PR has been reviewed.

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

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

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

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":["travis-minke-sap"]} -->
travis-minke-sap: This is currently just a POC for discussion / review...

/hold
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/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1384900534493777920/artifacts/line-cov.html#file0) | 82.5% | 87.7% | 5.1
[pkg/channel/distributed/dispatcher/dispatcher/handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1384900534493777920/artifacts/line-cov.html#file1) | 91.7% | 90.6% | -1.0

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/561?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#561](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/561?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (4152eb3) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/dcbe68cc1b0625fa16572097cc421ba3267e248f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (dcbe68c) will **decrease** coverage by `0.05%`.
> The diff coverage is `94.59%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/561/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/561?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     #561      +/-   ##
==========================================
- Coverage   71.68%   71.63%   -0.06%     
==========================================
  Files         137      137              
  Lines        5644     5619      -25     
==========================================
- Hits         4046     4025      -21     
+ Misses       1371     1368       -3     
+ Partials      227      226       -1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/561?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...annel/distributed/dispatcher/dispatcher/handler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/561/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2hhbmRsZXIuZ28=) | `86.95% <92.85%> (-0.55%)` | :arrow_down: |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/561/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `80.00% <100.00%> (-0.54%)` | :arrow_down: |

------

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

slinkydeveloper: /cc @aliok 
knative-prow-robot: @travis-minke-sap: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 4152eb3b939a471317d8af15ab0975ea1beb7002 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1384900534334394368) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/561). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-sap).

<details>

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

<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.
</details>
aliok: @travis-minke-sap  Sorry I missed this PR.

I think this is still a valid improvement. I would be happy to see this merged.  Thanks for the work.

I like the direction of the PR, not much to add at the moment. 
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/561#" title="Author self-approved">travis-minke-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)~~ [travis-minke-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":[]} -->
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/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1398340751191969792/artifacts/line-cov.html#file0) | 82.5% | 87.7% | 5.1
[pkg/channel/distributed/dispatcher/dispatcher/handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1398340751191969792/artifacts/line-cov.html#file1) | 93.0% | 92.3% | -0.7

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/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1400099307956211712/artifacts/line-cov.html#file0) | 82.5% | 87.7% | 5.1
[pkg/channel/distributed/dispatcher/dispatcher/handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1400099307956211712/artifacts/line-cov.html#file1) | 93.0% | 92.3% | -0.7

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/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1402280395768074240/artifacts/line-cov.html#file0) | 82.5% | 87.7% | 5.1
[pkg/channel/distributed/dispatcher/dispatcher/handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1402280395768074240/artifacts/line-cov.html#file1) | 93.0% | 92.3% | -0.7

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/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1402338990404669440/artifacts/line-cov.html#file0) | 82.5% | 87.2% | 4.7
[pkg/channel/distributed/dispatcher/dispatcher/handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1402338990404669440/artifacts/line-cov.html#file1) | 93.0% | 92.3% | -0.7

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/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1402351209397882880/artifacts/line-cov.html#file0) | 82.5% | 87.2% | 4.7
[pkg/channel/distributed/dispatcher/dispatcher/handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1402351209397882880/artifacts/line-cov.html#file1) | 93.0% | 92.3% | -0.7

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/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1403002344588709888/artifacts/line-cov.html#file0) | 82.2% | 87.1% | 4.9
[pkg/channel/distributed/dispatcher/dispatcher/handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/561/pull-knative-sandbox-eventing-kafka-go-coverage/1403002344588709888/artifacts/line-cov.html#file1) | 93.0% | 92.3% | -0.7

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/561#pullrequestreview-680930506" title="Approved">eric-sap</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/561#" title="Author self-approved">travis-minke-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)~~ [travis-minke-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":[]} -->
travis-minke-sap: /test pull-knative-sandbox-eventing-kafka-upgrade-tests

Patch:
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index 012014a32..e1dbf54fc 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -29,11 +29,11 @@ import (
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/channel"
 
-	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/consumer"
 	commonkafkautil "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/util"
 	dispatcherconstants "knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/constants"
 	"knative.dev/eventing-kafka/pkg/common/client"
 	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
+	commonconsumer "knative.dev/eventing-kafka/pkg/common/consumer"
 	"knative.dev/eventing-kafka/pkg/common/metrics"
 )
 
@@ -55,12 +55,11 @@ type SubscriberWrapper struct {
 	eventingduck.SubscriberSpec
 	GroupId       string
 	ConsumerGroup sarama.ConsumerGroup
-	StopChan      chan struct{}
 }
 
 // NewSubscriberWrapper Is The SubscriberWrapper Constructor
 func NewSubscriberWrapper(subscriberSpec eventingduck.SubscriberSpec, groupId string, consumerGroup sarama.ConsumerGroup) *SubscriberWrapper {
-	return &SubscriberWrapper{subscriberSpec, groupId, consumerGroup, make(chan struct{})}
+	return &SubscriberWrapper{subscriberSpec, groupId, consumerGroup}
 }
 
 // Dispatcher Interface
@@ -73,11 +72,12 @@ type Dispatcher interface {
 // DispatcherImpl Is A Struct With Configuration & ConsumerGroup State
 type DispatcherImpl struct {
 	DispatcherConfig
-	subscribers        map[types.UID]*SubscriberWrapper
-	consumerUpdateLock sync.Mutex
-	messageDispatcher  channel.MessageDispatcher
-	MetricsStopChan    chan struct{}
-	MetricsStoppedChan chan struct{}
+	subscribers          map[types.UID]*SubscriberWrapper
+	consumerGroupFactory commonconsumer.KafkaConsumerGroupFactory
+	consumerUpdateLock   sync.Mutex
+	messageDispatcher    channel.MessageDispatcher
+	MetricsStopChan      chan struct{}
+	MetricsStoppedChan   chan struct{}
 }
 
 // Verify The DispatcherImpl Implements The Dispatcher Interface
@@ -88,11 +88,12 @@ func NewDispatcher(dispatcherConfig DispatcherConfig) Dispatcher {
 
 	// Create The DispatcherImpl With Specified Configuration
 	dispatcher := &DispatcherImpl{
-		DispatcherConfig:   dispatcherConfig,
-		subscribers:        make(map[types.UID]*SubscriberWrapper),
-		messageDispatcher:  channel.NewMessageDispatcher(dispatcherConfig.Logger),
-		MetricsStopChan:    make(chan struct{}),
-		MetricsStoppedChan: make(chan struct{}),
+		DispatcherConfig:     dispatcherConfig,
+		subscribers:          make(map[types.UID]*SubscriberWrapper),
+		consumerGroupFactory: commonconsumer.NewConsumerGroupFactory(dispatcherConfig.Brokers, dispatcherConfig.SaramaConfig),
+		messageDispatcher:    channel.NewMessageDispatcher(dispatcherConfig.Logger),
+		MetricsStopChan:      make(chan struct{}),
+		MetricsStoppedChan:   make(chan struct{}),
 	}
 
 	// Start Observing Metrics
@@ -117,7 +118,7 @@ func (d *DispatcherImpl) Shutdown() {
 	}
 }
 
-// UpdateSubscriptions Updates The Dispatcher's Subscriptions To Align With New State
+// UpdateSubscriptions manages the Dispatcher's Subscriptions to align with new state
 func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error {
 
 	if d.SaramaConfig == nil {
@@ -145,8 +146,9 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 			// Create A ConsumerGroup Logger
 			logger := d.Logger.With(zap.String("GroupId", groupId))
 
-			// Attempt To Create A Kafka ConsumerGroup
-			consumerGroup, err := consumer.CreateConsumerGroup(d.Brokers, groupId, d.SaramaConfig)
+			// Create/Start A New ConsumerGroup With Custom Handler
+			handler := NewHandler(logger, groupId, &subscriberSpec)
+			consumerGroup, err := d.consumerGroupFactory.StartConsumerGroup(groupId, []string{d.Topic}, d.Logger.Sugar(), handler)
 			if err != nil {
 
 				// Log & Return Failure
@@ -158,8 +160,14 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 				// Create A New SubscriberWrapper With The ConsumerGroup
 				subscriber := NewSubscriberWrapper(subscriberSpec, groupId, consumerGroup)
 
-				// Start The ConsumerGroup Processing Messages
-				d.startConsuming(subscriber)
+				// Asynchronously Process ConsumerGroup's Error Channel
+				go func() {
+					logger.Info("ConsumerGroup Error Processing Initiated")
+					for err := range subscriber.ConsumerGroup.Errors() { // Closing ConsumerGroup Will Break Out Of This
+						logger.Error("ConsumerGroup Error", zap.Error(err))
+					}
+					logger.Info("ConsumerGroup Error Processing Terminated")
+				}()
 
 				// Track The New SubscriberWrapper For The SubscriberSpec As Active
 				d.subscribers[subscriberSpec.UID] = subscriber
@@ -190,59 +198,7 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 	return failedSubscriptions
 }
 
-// startConsuming Starts Consuming Messages With The Specified Subscriber's ConsumerGroup
-func (d *DispatcherImpl) startConsuming(subscriber *SubscriberWrapper) {
-
-	// Validate The Subscriber / ConsumerGroup
-	if subscriber != nil && subscriber.ConsumerGroup != nil {
-
-		// Setup The ConsumerGroup Level Logger
-		logger := d.Logger.With(zap.String("GroupId", subscriber.GroupId))
-
-		// Asynchronously Process ConsumerGroup's Error Channel
-		go func() {
-			logger.Info("ConsumerGroup Error Processing Initiated")
-			for err := range subscriber.ConsumerGroup.Errors() { // Closing ConsumerGroup Will Break Out Of This
-				logger.Error("ConsumerGroup Error", zap.Error(err))
-			}
-			logger.Info("ConsumerGroup Error Processing Terminated")
-		}()
-
-		// Create A New ConsumerGroupHandler To Consume Messages With
-		handler := NewHandler(logger, &subscriber.SubscriberSpec)
-
-		// Consume Messages Asynchronously
-		go func() {
-
-			// Infinite Loop To Support Server-Side ConsumerGroup Re-Balance Which Ends Consume() Execution
-			ctx := context.Background()
-			for {
-				select {
-
-				// Non-Blocking Stop Channel Check
-				case <-subscriber.StopChan:
-					logger.Info("ConsumerGroup Closed - Ceasing Consumption")
-					return
-
-				// Start ConsumerGroup Consumption
-				default:
-					logger.Info("ConsumerGroup Message Consumption Initiated")
-					err := subscriber.ConsumerGroup.Consume(ctx, []string{d.Topic}, handler)
-					if err != nil {
-						if err == sarama.ErrClosedConsumerGroup {
-							logger.Info("ConsumerGroup Closed Error - Ceasing Consumption") // Should be caught above but here as added precaution.
-							break
-						} else {
-							logger.Error("ConsumerGroup Failed To Consume Messages", zap.Error(err))
-						}
-					}
-				}
-			}
-		}()
-	}
-}
-
-// closeConsumerGroup Closes The ConsumerGroup Associated With A Single Subscriber
+// closeConsumerGroup closes the ConsumerGroup associated with a single Subscriber
 func (d *DispatcherImpl) closeConsumerGroup(subscriber *SubscriberWrapper) {
 
 	// Get The ConsumerGroup Associated with The Specified Subscriber
@@ -254,9 +210,6 @@ func (d *DispatcherImpl) closeConsumerGroup(subscriber *SubscriberWrapper) {
 	// If The ConsumerGroup Is Valid
 	if consumerGroup != nil {
 
-		// Mark The Subscriber's ConsumerGroup As Stopped
-		close(subscriber.StopChan)
-
 		// Close The ConsumerGroup
 		err := consumerGroup.Close()
 		if err != nil {
@@ -308,11 +261,9 @@ func (d *DispatcherImpl) SecretChanged(ctx context.Context, secret *corev1.Secre
 		return
 	}
 
-	// Create A New Dispatcher With The New Configuration (Reusing All Other Existing Config)
-	d.Logger.Info("Changes Detected In New Secret - Closing & Recreating Consumer Groups")
-
 	// Close all of the ConsumerGroups so that the settings can be changed
 	// Note: Not calling d.Shutdown because we don't want to close the metrics channel here
+	d.Logger.Info("Changes Detected In New Secret - Closing & Recreating Consumer Groups")
 	for _, subscriber := range d.subscribers {
 		d.closeConsumerGroup(subscriber)
 	}
@@ -321,6 +272,10 @@ func (d *DispatcherImpl) SecretChanged(ctx context.Context, secret *corev1.Secre
 	// of the SaramaConfig, so that's all that needs to be modified
 	d.DispatcherConfig.SaramaConfig = newConfig
 
+	// Replace The Dispatcher's ConsumerGroupFactory With Updated Version Using New Config
+	d.consumerGroupFactory = commonconsumer.NewConsumerGroupFactory(d.DispatcherConfig.Brokers, d.DispatcherConfig.SaramaConfig)
+
+	// Recreate the ConsumerGroups For Active Subscriptions
 	failedSubscriptions := d.UpdateSubscriptions(d.SubscriberSpecs)
 	if len(failedSubscriptions) > 0 {
 		d.Logger.Fatal("Failed To Subscribe Kafka Subscriptions Using Updated Secret", zap.Int("Count", len(failedSubscriptions)))
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/handler.go b/pkg/channel/distributed/dispatcher/dispatcher/handler.go
index 9fe1bb851..edd33bd2d 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/handler.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/handler.go
@@ -27,112 +27,92 @@ import (
 	"github.com/cloudevents/sdk-go/v2/binding"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
-	kafkasarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
-	"knative.dev/eventing-kafka/pkg/common/tracing"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/channel"
 	"knative.dev/eventing/pkg/kncloudevents"
+
+	commonconsumer "knative.dev/eventing-kafka/pkg/common/consumer"
+	kafkasarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
+	"knative.dev/eventing-kafka/pkg/common/tracing"
 )
 
-// Verify The Handler Implements The Sarama ConsumerGroupHandler
-var _ sarama.ConsumerGroupHandler = &Handler{}
+// Verify The Handler Implements The Common KafkaConsumerHandler
+var _ commonconsumer.KafkaConsumerHandler = &Handler{}
 
-// Handler Defines A Sarama ConsumerGroupHandler Implementation
+// Handler Struct implementing the KafkaConsumerHandler Interface
 type Handler struct {
 	Logger            *zap.Logger
+	GroupId           string
 	Subscriber        *eventingduck.SubscriberSpec
 	MessageDispatcher channel.MessageDispatcher
+	destinationURL    *url.URL
+	replyURL          *url.URL
+	deadLetterURL     *url.URL
+	retryConfig       kncloudevents.RetryConfig
 }
 
-// NewHandler Creates A New Handler
-func NewHandler(logger *zap.Logger, subscriber *eventingduck.SubscriberSpec) *Handler {
-	return &Handler{
+// NewHandler creates a new Handler instance.
+func NewHandler(logger *zap.Logger, groupId string, subscriber *eventingduck.SubscriberSpec) *Handler {
+
+	// Create The New Handler Instance
+	handler := &Handler{
 		Logger:            logger,
+		GroupId:           groupId,
 		Subscriber:        subscriber,
 		MessageDispatcher: newMessageDispatcherWrapper(logger),
 	}
-}
-
-// Wrapper Function To Facilitate Testing With A Mock Knative MessageDispatcher
-var newMessageDispatcherWrapper = func(logger *zap.Logger) channel.MessageDispatcher {
-	return channel.NewMessageDispatcher(logger)
-}
-
-// Setup Is The Initial ConsumerGroupHandler Lifecycle Method (Runs before any ConsumeClaims)
-func (h *Handler) Setup(_ sarama.ConsumerGroupSession) error {
-	return nil // Nothing To Do As Of Yet
-}
-
-// Cleanup Is The Final ConsumerGroupHandler Lifecycle Method (Runs after all ConsumeClaims stop but before final offset commit)
-func (h *Handler) Cleanup(_ sarama.ConsumerGroupSession) error {
-	return nil // Nothing To Do As Of Yet
-}
-
-// ConsumeClaim Is The Main ConsumerGroupHandler Lifecycle Method (Main processing loop, must finish when claim.Messages() channel closes.)
-func (h *Handler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
 
 	// Extract The Destination URL From The Subscriber
-	var destinationURL *url.URL
-	if !h.Subscriber.SubscriberURI.IsEmpty() {
-		destinationURL = h.Subscriber.SubscriberURI.URL()
+	if !subscriber.SubscriberURI.IsEmpty() {
+		handler.destinationURL = subscriber.SubscriberURI.URL()
 	}
 
 	// Extract The Reply URL From The Subscriber
-	var replyURL *url.URL
-	if !h.Subscriber.ReplyURI.IsEmpty() {
-		replyURL = h.Subscriber.ReplyURI.URL()
+	if !subscriber.ReplyURI.IsEmpty() {
+		handler.replyURL = subscriber.ReplyURI.URL()
 	}
 
-	// Validate The Subscriber's Delivery (Optional)
-	var deadLetterURL *url.URL
-	retryConfig := kncloudevents.NoRetries()
-	if h.Subscriber.Delivery != nil {
+	// Validate The Subscriber's Delivery (Optional - Defaults To No Retries Or DLQ)
+	handler.retryConfig = kncloudevents.NoRetries()
+	if subscriber.Delivery != nil {
 
 		// Extract The DeadLetterSink From The Subscriber.Delivery
-		if h.Subscriber.Delivery.DeadLetterSink != nil &&
-			h.Subscriber.Delivery.DeadLetterSink.URI != nil &&
-			!h.Subscriber.Delivery.DeadLetterSink.URI.IsEmpty() {
-			deadLetterURL = h.Subscriber.Delivery.DeadLetterSink.URI.URL()
+		if subscriber.Delivery.DeadLetterSink != nil &&
+			subscriber.Delivery.DeadLetterSink.URI != nil &&
+			!subscriber.Delivery.DeadLetterSink.URI.IsEmpty() {
+			handler.deadLetterURL = subscriber.Delivery.DeadLetterSink.URI.URL()
 		}
 
-		// Extract The RetryConfig From The Subscriber.Delivery (Defaults To NoRetries)
+		// Extract The RetryConfig From The Subscriber.Delivery
 		var err error
-		retryConfig, err = kncloudevents.RetryConfigFromDeliverySpec(*h.Subscriber.Delivery)
+		handler.retryConfig, err = kncloudevents.RetryConfigFromDeliverySpec(*subscriber.Delivery)
 		if err != nil {
-			h.Logger.Error("Failed To Parse RetryConfig From DeliverySpec - No Retries Will Occur", zap.Error(err))
+			logger.Error("Failed To Parse RetryConfig From DeliverySpec - No Retries Will Occur", zap.Error(err))
 		} else {
-			h.Logger.Info("Successfully Parsed RetryConfig From DeliverySpec", zap.Int("RetryMax", retryConfig.RetryMax))
-			retryConfig.CheckRetry = kncloudevents.SelectiveRetry // Specify Custom CheckRetry Function
+			logger.Info("Successfully Parsed RetryConfig From DeliverySpec", zap.Int("RetryMax", handler.retryConfig.RetryMax))
+			handler.retryConfig.CheckRetry = kncloudevents.SelectiveRetry // Specify Custom CheckRetry Function
 		}
 	}
 
-	// Pull Any Available Messages From The ConsumerGroupClaim (Until The Channel Closes)
-	for message := range claim.Messages() {
-
-		// Consume The Message (Ignore any errors other than "context canceled" - Will have already been retried
-		// and we're moving on so as not to block further Topic processing.)
-		err := h.consumeMessage(session.Context(), message, destinationURL, replyURL, deadLetterURL, &retryConfig)
-
-		// If the system is shutting down, it's possible to get a "unable to complete request to [destination]: context canceled" error
-		// This is different than other types of failure in that it is the dispatcher's "fault" and so marking the message as delivered
-		// would be a violation of the "at least once" guarantee (because the next dispatcher that starts would see it as marked and not
-		// try to deliver it again).
-		if err == nil || !strings.Contains(err.Error(), context.Canceled.Error()) {
-			// Mark The Message As Having Been Consumed (Does Not Imply Successful Delivery - Only Full Retry Attempts Made)
-			session.MarkMessage(message, "")
-		}
-	}
+	// Return The Configured Handler
+	return handler
+}
 
-	// Return Success
-	return nil
+// Wrapper Function To Facilitate Testing With A Mock Knative MessageDispatcher
+var newMessageDispatcherWrapper = func(logger *zap.Logger) channel.MessageDispatcher {
+	return channel.NewMessageDispatcher(logger)
 }
 
-// consumeMessage Consumes A Single Message
-func (h *Handler) consumeMessage(context context.Context, consumerMessage *sarama.ConsumerMessage, destinationURL *url.URL, replyURL *url.URL, deadLetterURL *url.URL, retryConfig *kncloudevents.RetryConfig) error {
+// Handle is responsible for processing the individual ConsumerMessages.  The
+// first return bool indicates whether to MarkOffset in the ConsumerGroup and
+// the second error value will be sent to the ConsumerGroups error channel as
+// well as the SetReady() function.
+func (h *Handler) Handle(ctx context.Context, consumerMessage *sarama.ConsumerMessage) (bool, error) {
 
-	// Debug Log Kafka ConsumerMessage
+	// Debug Log Kafka ConsumerMessage (Verify Debug Level For Efficiency!)
 	if h.Logger.Core().Enabled(zap.DebugLevel) {
-		// Checked Logging Level First To Avoid Calling StringifyHeaderPtrs In Production
+
+		// Checked Logging Level First To Avoid Calling StringifyHeaderPtrs() In Production
 		h.Logger.Debug("Consuming Kafka Message",
 			zap.Any("Headers", kafkasarama.StringifyHeaderPtrs(consumerMessage.Headers)), // Log human-readable strings, not base64
 			zap.ByteString("Key", consumerMessage.Key),
@@ -146,17 +126,61 @@ func (h *Handler) consumeMessage(context context.Context, consumerMessage *saram
 	message := kafkasaramaprotocol.NewMessageFromConsumerMessage(consumerMessage)
 	if message.ReadEncoding() == binding.EncodingUnknown {
 		h.Logger.Warn("Received A Message With Unknown Encoding - Skipping")
-		return errors.New("received a message with unknown encoding - skipping")
+		return true, errors.New("received a message with unknown encoding - skipping") // Mark As Handled Since Retry Won't Fix Anything : )
 	}
 
-	ctx, span := tracing.StartTraceFromMessage(h.Logger.Sugar(), context, message, consumerMessage.Topic)
+	// Start Tracing
+	ctx, span := tracing.StartTraceFromMessage(h.Logger.Sugar(), ctx, message, consumerMessage.Topic)
 	defer span.End()
 
-	// Dispatch The Message With Configured Retries & Return Any Errors
-	info, dispatchError := h.MessageDispatcher.DispatchMessageWithRetries(ctx, message, nil, destinationURL, replyURL, deadLetterURL, retryConfig)
+	// Dispatch The Message With Configured Retries, DLQ, etc
+	info, err := h.MessageDispatcher.DispatchMessageWithRetries(ctx, message, nil, h.destinationURL, h.replyURL, h.deadLetterURL, &h.retryConfig)
 	h.Logger.Debug("Received Response", zap.Any("ExecutionInfo", executionInfoWrapper{info}))
 
-	return dispatchError
+	//
+	// Determine Whether To Mark The Message As Processed
+	// (Does Not Imply Successful Delivery - Only Full Retry Attempts Made)
+	//
+	// In general all messages are marked as Processed, even if the result of
+	// processing was not a successful delivery. This is to prevent blocking
+	// the Dispatcher forever on invalid messages, or faulty subscribers, etc.
+	// If, however, the Dispatcher is shutting down it is possible to get an
+	// "unable to complete request to [destination]: context canceled" error
+	// for which the message might not have been given a fair chance and needs
+	// to be attempted again upon subsequent restart.
+	//
+	// This is different from the Consolidated KafkaChannel implementation
+	// which only returns true if message was delivered successfully.
+	//
+	markMessage := true
+	if err != nil && strings.Contains(err.Error(), context.Canceled.Error()) {
+		markMessage = false
+	}
+
+	//
+	// Return The Results Of Handling The ConsumerMessage
+	//
+	// Errors are not currently being returned in order to maintain consistent
+	// behavior with the implementation prior to using the common/consumer
+	// KafkaConsumerGroupFactory.  We do NOT want to mark the status of the
+	// ConsumerGroup as not-ready just because a single messages encountered an
+	// error.  This is an important and conscious design choice which enables
+	// the use of multi-tenant KafkaChannels and other advanced use cases.
+	//
+	return markMessage, nil
+}
+
+// SetReady is used by the "Prober" implementation for tracking ConsumerGroup
+// status which we are not using at the moment, and is believed to be
+// undergoing refactor / replacement in favor of using the control-protocol
+// and is thus not supported here yet ; )
+func (h *Handler) SetReady(partition int32, ready bool) {
+	h.Logger.Debug("No-Op SetReady Handler", zap.Int32("Partition", partition), zap.Bool("Ready", ready))
+}
+
+// GetConsumerGroup returns the ConsumerGroup ID of the Handler
+func (h *Handler) GetConsumerGroup() string {
+	return h.GroupId
 }
 
 // executionInfoWrapper wraps a DispatchExecutionInfo struct so that zap.Any can lazily marshal it
diff --git a/pkg/channel/distributed/receiver/README.md b/pkg/channel/distributed/receiver/README.md
index c7502fd37..2f49405be 100644
--- a/pkg/channel/distributed/receiver/README.md
+++ b/pkg/channel/distributed/receiver/README.md
@@ -4,6 +4,8 @@ The Receiver implementation is a Kafka Producer which is responsible for
 receiving CloudEvents, converting them to Kafka Messages, and writing them to
 the appropriate Kafka Topic.
 
+## Runtime Deployments
+
 A unique Deployment / Service is created for every installation. The single
 Deployment is horizontally scalable as necessary; the replicas will initially be
 configured by the eventing-kafka.receiver.replicas field in the
@@ -22,8 +24,33 @@ to the appropriate Kafka Topic.
 The Kafka brokers and credentials are obtained from mounted data in the Kafka
 Secret (by default named `kafka-cluster` in the knative-eventing namespace,
 configurable by altering the kafka.authSecretName and/or the
-kafka.authSecretNamespace in the
-[ConfigMap](../../../../config/channel/distributed/300-eventing-kafka-configmap.yaml)).
+kafka.authSecretNamespace values in the
+[ConfigMap](../../../../config/channel/distributed/300-eventing-kafka-configmap.yaml))
+.
+
+## CPU Requirements
+
+Providing CPU guidance is a difficult endeavor as there are so many variables
+(event size, event rate, acceptable latency, kafka provider, use of an event
+mesh, etc...) that can affect performance. The following is a single data point
+to help you get started, and you should analyze your particular use case to fine
+tune.
+
+Sending ~2000 events/second on a single replica of the receiver, without an
+event-mesh, where each event is approximately 20Kb in size, required **~1000m**
+(a full CPU core) where the CPU was a GCP
+[N1-Standard-2](https://cloud.google.com/compute/docs/cpu-platforms) 2 Core
+Intel Skylake processor with a base clock of 2.0GHz
+
+## Memory Requirements
+
+The receiver simply verifies the destination KafkaChannel (Topic) and produces
+the event. It does very little with the actual CloudEvent and thus the memory
+usage is rather low as the events are transitory and garbage collected.
+Preliminary performance testing has shown that the receiver will require less
+than **~50Mi** to process ~2000 events/second where each event is approximately
+20Kb in size. These are rough guidelines, and you should perform your own
+analysis, but hopefully this provides a useful starting point.
 
 ## Tracing, Profiling, and Metrics
 
diff --git a/pkg/common/client/config.go b/pkg/common/client/config.go
index 7ff20dd84..caf3359bb 100644
--- a/pkg/common/client/config.go
+++ b/pkg/common/client/config.go
@@ -253,7 +253,6 @@ func (b *configBuilder) Build(ctx context.Context) (*sarama.Config, error) {
 				config.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
 			}
 			config.Net.SASL.User = b.auth.SASL.User
-			config.Net.SASL.Password = b.auth.SASL.Password
 		}
 	}
 
@@ -268,6 +267,10 @@ func (b *configBuilder) Build(ctx context.Context) (*sarama.Config, error) {
 	logger := logging.FromContext(ctx)
 	logger.Infof("Built Sarama config: %+v", config)
 
+	if b.auth != nil && b.auth.SASL != nil {
+		config.Net.SASL.Password = b.auth.SASL.Password
+	}
+
 	return config, nil
 }
 


Test patch:
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
index 283713074..b5f24eff7 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
@@ -27,16 +27,16 @@ import (
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
-	consumertesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/consumer/testing"
-	consumerwrapper "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/consumer/wrapper"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/pkg/logging"
+	logtesting "knative.dev/pkg/logging/testing"
+
 	commonclient "knative.dev/eventing-kafka/pkg/common/client"
 	clienttesting "knative.dev/eventing-kafka/pkg/common/client/testing"
 	configtesting "knative.dev/eventing-kafka/pkg/common/config/testing"
+	consumertesting "knative.dev/eventing-kafka/pkg/common/consumer/testing"
 	"knative.dev/eventing-kafka/pkg/common/metrics"
 	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
-	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
-	"knative.dev/pkg/logging"
-	logtesting "knative.dev/pkg/logging/testing"
 )
 
 // Test Data
@@ -65,7 +65,6 @@ func TestNewSubscriberWrapper(t *testing.T) {
 	assert.Equal(t, subscriber.UID, subscriberWrapper.UID)
 	assert.Equal(t, consumerGroup, subscriberWrapper.ConsumerGroup)
 	assert.Equal(t, groupId, subscriberWrapper.GroupId)
-	assert.NotNil(t, subscriberWrapper.StopChan)
 }
 
 // Test The NewDispatcher() Functionality
@@ -128,9 +127,6 @@ func TestUpdateSubscriptions(t *testing.T) {
 	logger := logtesting.TestLogger(t)
 	ctx := logging.WithLogger(context.TODO(), logger)
 
-	// Restore Any Stubbing Of NewConsumerGroupWrapper After Test Is Finished
-	defer consumertesting.RestoreNewConsumerGroupFn()
-
 	// Test Data
 	brokers := []string{configtesting.DefaultKafkaBroker}
 	config, err := commonclient.NewConfigBuilder().WithDefaults().FromYaml(clienttesting.DefaultSaramaConfigYaml).Build(ctx)
@@ -254,18 +250,11 @@ func TestUpdateSubscriptions(t *testing.T) {
 	for _, testCase := range filteredTestCases {
 		t.Run(testCase.name, func(t *testing.T) {
 
-			// Mock The ConsumerGroup & Stub The NewSyncConsumerGroupWrapper()
-			mockConsumerGroup := consumertesting.NewMockConsumerGroup()
-			consumertesting.StubNewConsumerGroupFn(customValidationNewConsumerGroupFn(t,
-				testCase.fields.DispatcherConfig.Brokers,
-				testCase.args.subscriberSpecs,
-				testCase.fields.DispatcherConfig.SaramaConfig,
-				mockConsumerGroup))
-
 			// Create A New DispatcherImpl To Test
 			dispatcher := &DispatcherImpl{
-				DispatcherConfig: testCase.fields.DispatcherConfig,
-				subscribers:      testCase.fields.subscribers,
+				DispatcherConfig:     testCase.fields.DispatcherConfig,
+				subscribers:          testCase.fields.subscribers,
+				consumerGroupFactory: &consumertesting.MockKafkaConsumerGroupFactory{},
 			}
 
 			// Perform The Test
@@ -362,17 +351,10 @@ func TestSecretChanged(t *testing.T) {
 		},
 	}
 
-	// Make Sure To Restore The NewConsumerGroup Wrapper After The Test
-	defer consumertesting.RestoreNewConsumerGroupFn()
-
 	// Run The Filtered TestCases
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
 
-			// Mock The SyncProducer & Stub The NewConsumerGroupWrapper()
-			mockConsumerGroup := consumertesting.NewMockConsumerGroup()
-			consumertesting.StubNewConsumerGroupFn(consumertesting.NonValidatingNewConsumerGroupFn(mockConsumerGroup))
-
 			// Create A Test Dispatcher To Perform Tests Against
 			dispatcher := createTestDispatcher(t, brokers, baseSaramaConfig)
 			impl := dispatcher.(*DispatcherImpl)
@@ -437,26 +419,6 @@ func createTestDispatcher(t *testing.T, brokers []string, config *sarama.Config)
 	return dispatcher
 }
 
-// Custom Validation Function For NewConsumerGroup - Capable Of Verifying Multiple GroupIds
-func customValidationNewConsumerGroupFn(t *testing.T,
-	expectedBrokers []string,
-	subscriberSpecs []eventingduck.SubscriberSpec,
-	expectedConfig *sarama.Config,
-	mockConsumerGroup sarama.ConsumerGroup) consumerwrapper.NewConsumerGroupFnType {
-
-	expectedGroupIds := make([]string, len(subscriberSpecs))
-	for _, subscriberSpec := range subscriberSpecs {
-		expectedGroupIds = append(expectedGroupIds, fmt.Sprintf("kafka.%s", subscriberSpec.UID))
-	}
-
-	return func(brokers []string, groupId string, config *sarama.Config) (sarama.ConsumerGroup, error) {
-		assert.Equal(t, brokers, expectedBrokers)
-		assert.Contains(t, expectedGroupIds, groupId)
-		assert.Equal(t, config, expectedConfig)
-		return mockConsumerGroup, nil
-	}
-}
-
 func TestConfigImpl_ObserveMetrics(t *testing.T) {
 	baseSaramaConfig, err := commonclient.NewConfigBuilder().
 		WithDefaults().
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/handler_test.go b/pkg/channel/distributed/dispatcher/dispatcher/handler_test.go
index bf5d8ad62..70767aad3 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/handler_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/handler_test.go
@@ -1,12 +1,9 @@
 /*
 Copyright 2020 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.
@@ -18,6 +15,7 @@ package dispatcher
 
 import (
 	"context"
+	"fmt"
 	"net/url"
 	"strings"
 	"testing"
@@ -67,6 +65,7 @@ var (
 	testBackoffDelay     = "PT1S"
 	testDeadLetterURI, _ = apis.ParseURL(testDeadLetterURIString)
 	testMsgTime          = time.Now().UTC().Format(time.RFC3339)
+	testConsumerGroupId  = fmt.Sprintf("kafka.%s", testSubscriberUID)
 )
 
 // Test The NewHandler() Functionality
@@ -74,83 +73,77 @@ func TestNewHandler(t *testing.T) {
 	assert.NotNil(t, createTestHandler)
 }
 
-// Test The Handler's Setup() Functionality
-func TestHandlerSetup(t *testing.T) {
-	handler := createTestHandler(t, testSubscriberURI, testReplyURI, nil)
-	assert.Nil(t, handler.Setup(nil))
-}
-
-// Test The Handler's Cleanup() Functionality
-func TestHandlerCleanup(t *testing.T) {
-	handler := createTestHandler(t, testSubscriberURI, testReplyURI, nil)
-	assert.Nil(t, handler.Cleanup(nil))
-}
-
-type HandlerConsumeClaimTestCase struct {
-	only           bool
-	name           string
-	destinationUri *apis.URL
-	replyUri       *apis.URL
-	deadLetterUri  *apis.URL
-	dispatchErr    error
-	retry          bool
-	expectSkipMark bool
+type HandleTestCase struct {
+	only              bool
+	name              string
+	destinationUri    *apis.URL
+	replyUri          *apis.URL
+	deadLetterUri     *apis.URL
+	dispatchErr       error
+	retry             bool
+	expectMarkMessage bool
 }
 
-// Test The Handler's ConsumeClaim() Functionality
-func TestHandlerConsumeClaim(t *testing.T) {
+// Test The Handler's Handle() Functionality
+func TestHandle(t *testing.T) {
 
-	// Define The HandlerConsumeClaimTestCases
-	testCases := []HandlerConsumeClaimTestCase{
+	// Define The HandleTestCases
+	testCases := []HandleTestCase{
 		{
-			name:           "Complete Subscriber Configuration",
-			destinationUri: testSubscriberURI,
-			replyUri:       testReplyURI,
-			deadLetterUri:  testDeadLetterURI,
-			retry:          true,
+			name:              "Complete Subscriber Configuration",
+			destinationUri:    testSubscriberURI,
+			replyUri:          testReplyURI,
+			deadLetterUri:     testDeadLetterURI,
+			retry:             true,
+			expectMarkMessage: true,
 		},
 		{
-			name:          "No Subscriber URL",
-			replyUri:      testReplyURI,
-			deadLetterUri: testDeadLetterURI,
-			retry:         true,
+			name:              "No Subscriber URL",
+			replyUri:          testReplyURI,
+			deadLetterUri:     testDeadLetterURI,
+			retry:             true,
+			expectMarkMessage: true,
 		},
 		{
-			name:           "No Reply URL",
-			destinationUri: testSubscriberURI,
-			deadLetterUri:  testDeadLetterURI,
-			retry:          true,
+			name:              "No Reply URL",
+			destinationUri:    testSubscriberURI,
+			deadLetterUri:     testDeadLetterURI,
+			retry:             true,
+			expectMarkMessage: true,
 		},
 		{
-			name:           "No DeadLetter URL",
-			destinationUri: testSubscriberURI,
-			replyUri:       testReplyURI,
-			retry:          true,
+			name:              "No DeadLetter URL",
+			destinationUri:    testSubscriberURI,
+			replyUri:          testReplyURI,
+			retry:             true,
+			expectMarkMessage: true,
 		},
 		{
-			name:           "No Retry",
-			destinationUri: testSubscriberURI,
-			replyUri:       testReplyURI,
-			deadLetterUri:  testDeadLetterURI,
-			retry:          false,
+			name:              "No Retry",
+			destinationUri:    testSubscriberURI,
+			replyUri:          testReplyURI,
+			deadLetterUri:     testDeadLetterURI,
+			retry:             false,
+			expectMarkMessage: true,
 		},
 		{
-			name:  "Empty Subscriber Configuration",
-			retry: false,
+			name:              "Empty Subscriber Configuration",
+			retry:             false,
+			expectMarkMessage: true,
 		},
 		{
-			name:           "Context Canceled",
-			destinationUri: testSubscriberURI,
-			replyUri:       testReplyURI,
-			deadLetterUri:  testDeadLetterURI,
-			retry:          true,
-			dispatchErr:    context.Canceled,
-			expectSkipMark: true,
+			name:              "Context Canceled",
+			destinationUri:    testSubscriberURI,
+			replyUri:          testReplyURI,
+			deadLetterUri:     testDeadLetterURI,
+			retry:             true,
+			dispatchErr:       context.Canceled,
+			expectMarkMessage: false,
 		},
 	}
 
 	// Filter To Those With "only" Flag (If Any Specified)
-	filteredTestCases := make([]HandlerConsumeClaimTestCase, 0)
+	filteredTestCases := make([]HandleTestCase, 0)
 	for _, testCase := range testCases {
 		if testCase.only {
 			filteredTestCases = append(filteredTestCases, testCase)
@@ -163,13 +156,24 @@ func TestHandlerConsumeClaim(t *testing.T) {
 	// Execute The Individual Test Cases
 	for _, testCase := range filteredTestCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			performHandlerConsumeClaimTest(t, testCase)
+			performHandleTest(t, testCase)
 		})
 	}
 }
 
-// Test One Permutation Of The Handler's ConsumeClaim() Functionality
-func performHandlerConsumeClaimTest(t *testing.T, testCase HandlerConsumeClaimTestCase) {
+func TestSetReady(t *testing.T) {
+	handler := createTestHandler(t, testSubscriberURI, testReplyURI, nil)
+	handler.SetReady(1, true)
+}
+
+func TestGetConsumerGroup(t *testing.T) {
+	handler := createTestHandler(t, testSubscriberURI, testReplyURI, nil)
+	actualConsumerGroupId := handler.GetConsumerGroup()
+	assert.Equal(t, testConsumerGroupId, actualConsumerGroupId)
+}
+
+// Test One Permutation Of The Handler's Handle() Functionality
+func performHandleTest(t *testing.T, testCase HandleTestCase) {
 
 	// Initialize Destination As Specified
 	var destinationUrl *url.URL
@@ -202,8 +206,6 @@ func performHandlerConsumeClaimTest(t *testing.T, testCase HandlerConsumeClaimTe
 	}
 
 	// Create Mocks For Testing
-	mockConsumerGroupSession := dispatchertesting.NewMockConsumerGroupSession(t)
-	mockConsumerGroupClaim := dispatchertesting.NewMockConsumerGroupClaim(t)
 	mockMessageDispatcher := dispatchertesting.NewMockMessageDispatcher(t, nil, destinationUrl, replyUrl, deadLetterUrl, &retryConfig, testCase.dispatchErr)
 
 	// Mock The newMessageDispatcherWrapper Function (And Restore Post-Test)
@@ -216,34 +218,15 @@ func performHandlerConsumeClaimTest(t *testing.T, testCase HandlerConsumeClaimTe
 	// Create The Handler To Test
 	handler := createTestHandler(t, testCase.destinationUri, testCase.replyUri, &deliverySpec)
 
-	consumeFinishedChan := make(chan bool)
-	// Background Start Consuming Claims
-	go func() {
-		err := handler.ConsumeClaim(mockConsumerGroupSession, mockConsumerGroupClaim)
-		assert.Nil(t, err)
-		consumeFinishedChan <- true
-	}()
-
-	// Perform The Test (Add ConsumerMessages To Claims)
+	// Perform The Test
 	consumerMessage := createConsumerMessage(t)
-	mockConsumerGroupClaim.MessageChan <- consumerMessage
-
-	// Verify The Results (CloudEvent Was Dispatched & ConsumerMessage Was Marked)
-	if !testCase.expectSkipMark {
-		// Wait For Message To Be Marked As Complete
-		markedMessage := <-mockConsumerGroupSession.MarkMessageChan
+	result, err := handler.Handle(context.TODO(), consumerMessage)
 
-		assert.Equal(t, consumerMessage, markedMessage)
-		assert.NotNil(t, mockMessageDispatcher.Message())
-		verifyDispatchedMessage(t, mockMessageDispatcher.Message())
-	}
-	assert.Equal(t, !testCase.expectSkipMark, mockConsumerGroupSession.MarkMessageCalled())
-
-	// Close The Mock ConsumerGroupClaim Message Channel To Complete/Exit Handler's ConsumeClaim()
-	close(mockConsumerGroupClaim.MessageChan)
-
-	// Wait for the consumer goroutine to finish to prevent logging attempts after the test is completed
-	<-consumeFinishedChan
+	// Verify The Results
+	assert.Nil(t, err)
+	assert.Equal(t, testCase.expectMarkMessage, result)
+	assert.NotNil(t, mockMessageDispatcher.Message())
+	verifyDispatchedMessage(t, mockMessageDispatcher.Message())
 }
 
 // Verify The Dispatched Message Contains Test Message Contents (Was Not Corrupted)
@@ -303,7 +286,7 @@ func createTestHandler(t *testing.T, subscriberURL *apis.URL, replyUrl *apis.URL
 	}
 
 	// Perform The Test Create The Test Handler
-	handler := NewHandler(logger, testSubscriber)
+	handler := NewHandler(logger, testConsumerGroupId, testSubscriber)
 
 	// Verify The Results
 	assert.NotNil(t, handler)
diff --git a/pkg/common/consumer/testing/mocks.go b/pkg/common/consumer/testing/mocks.go
new file mode 100644
index 000000000..4fc874e11
--- /dev/null
+++ b/pkg/common/consumer/testing/mocks.go
@@ -0,0 +1,81 @@
+/*
+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 testing
+
+import (
+	"context"
+	"errors"
+
+	"github.com/Shopify/sarama"
+	"go.uber.org/zap"
+
+	"knative.dev/eventing-kafka/pkg/common/consumer"
+)
+
+//
+// Mock KafkaConsumerGroupFactory
+//
+
+type MockKafkaConsumerGroupFactory struct {
+	// CreateErr will return an error when creating a consumer
+	CreateErr bool
+}
+
+func (c MockKafkaConsumerGroupFactory) StartConsumerGroup(groupID string, topics []string, logger *zap.SugaredLogger, handler consumer.KafkaConsumerHandler, options ...consumer.SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error) {
+	if c.CreateErr {
+		return nil, errors.New("error creating consumer")
+	}
+	return NewMockConsumerGroup(), nil
+}
+
+var _ consumer.KafkaConsumerGroupFactory = (*MockKafkaConsumerGroupFactory)(nil)
+
+//
+// Mock ConsumerGroup
+//
+
+var _ sarama.ConsumerGroup = &MockConsumerGroup{}
+
+type MockConsumerGroup struct {
+	errorChan   chan error
+	consumeChan chan struct{}
+	Closed      bool
+}
+
+func NewMockConsumerGroup() *MockConsumerGroup {
+	return &MockConsumerGroup{
+		errorChan:   make(chan error),
+		consumeChan: make(chan struct{}),
+		Closed:      false,
+	}
+}
+
+func (m *MockConsumerGroup) Consume(_ context.Context, _ []string, _ sarama.ConsumerGroupHandler) error {
+	<-m.consumeChan                      // Block To Simulate Real Execution
+	return sarama.ErrClosedConsumerGroup // Return ConsumerGroup Closed "Error" For Clean Shutdown
+}
+
+func (m *MockConsumerGroup) Errors() <-chan error {
+	return m.errorChan
+}
+
+func (m *MockConsumerGroup) Close() error {
+	close(m.errorChan)
+	close(m.consumeChan)
+	m.Closed = true
+	return nil
+}