Repository: knative-sandbox/eventing-kafka
Pull request #741 (2021-06-25 21:57:10)

Resetoffset Controller - Reconciler Lifecycle & Locks
Subtask of #588 to add locks to ConsumerGroupAsyncCommands and refactor reconciler to retry on failure.

## Proposed Changes

- 🧽  Cleanup prior TODOs from #725 now that #724 is merged.
- 🧽  Separate util functions for easier testing.
- 🎁  Add lock fields to Async ConsumerGroup messages (not yet supported by Consumer Manager but non-breaking)
- 🎁. Refactor ResetOffset reconciler to retry on error with appropriate handling for various levels of completion.
- 🎁  Adjust ResetOffset Status/Lifecycle to align with new approach.

**Notes**
- The ResetOffset Controller is still "inactive" and has not been added to any Channels/Sources/etc.
- The ResetOffset Controller is still not "complete" and this is a stepping stone in the path to implementing the feature. 
- A forthcoming PR will implement the actual "locking" of ConsumerGroupId's in the Consumer Manager, this PR only sends appropriate lock tokens to prevent overlapping reconciliation.


**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/741#" 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/common/commands/resetoffset/controller/dataplane.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/741/pull-knative-sandbox-eventing-kafka-go-coverage/1408544833286443008/artifacts/line-cov.html#file2) | 85.9% | 84.8% | -1.1
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/741/pull-knative-sandbox-eventing-kafka-go-coverage/1408544833286443008/artifacts/line-cov.html#file3) | 97.6% | 97.6% | -0.0
[pkg/common/commands/resetoffset/controller/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/741/pull-knative-sandbox-eventing-kafka-go-coverage/1408544833286443008/artifacts/line-cov.html#file4) | Do not exist | 83.3% | 
[pkg/common/controlprotocol/commands/lock.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/741/pull-knative-sandbox-eventing-kafka-go-coverage/1408544833286443008/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#741](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (57d9100) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/000dd9c4955f4b3c3e2f71b35d30138c7d1b9770?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (000dd9c) will **increase** coverage by `0.05%`.
> The diff coverage is `75.88%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741/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/741?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     #741      +/-   ##
==========================================
+ Coverage   75.29%   75.34%   +0.05%     
==========================================
  Files         132      140       +8     
  Lines        5785     6231     +446     
==========================================
+ Hits         4356     4695     +339     
- Misses       1208     1316     +108     
+ Partials      221      220       -1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...onsolidated/dispatcher/consumer\_message\_handler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvY29uc3VtZXJfbWVzc2FnZV9oYW5kbGVyLmdv) | `0.00% <0.00%> (ø)` | |
| [.../channel/distributed/controller/util/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741/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/source/adapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9hZGFwdGVyL2FkYXB0ZXIuZ28=) | `58.42% <ø> (-1.08%)` | :arrow_down: |
| [pkg/source/client/client.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9jbGllbnQvY2xpZW50Lmdv) | `15.66% <0.00%> (-19.48%)` | :arrow_down: |
| [pkg/source/reconciler/source/kafkasource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9yZWNvbmNpbGVyL3NvdXJjZS9rYWZrYXNvdXJjZS5nbw==) | `0.00% <0.00%> (ø)` | |
| [pkg/source/mtadapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvYWRhcHRlci5nbw==) | `47.82% <22.72%> (-27.36%)` | :arrow_down: |
| [pkg/apis/sources/v1beta1/kafka\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMvc291cmNlcy92MWJldGExL2thZmthX2xpZmVjeWNsZS5nbw==) | `53.33% <50.00%> (-0.52%)` | :arrow_down: |
| [pkg/channel/consolidated/utils/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3V0aWxzL3V0aWwuZ28=) | `95.91% <50.00%> (-4.09%)` | :arrow_down: |
| [pkg/source/client/offsets.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9jbGllbnQvb2Zmc2V0cy5nbw==) | `60.00% <60.00%> (ø)` | |
| [pkg/channel/consolidated/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvZGlzcGF0Y2hlci5nbw==) | `61.58% <66.66%> (+0.10%)` | :arrow_up: |
| ... and [31 more](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741/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/741?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/741?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [8d6dae8...57d9100](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/741?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).

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-upgrade-tests | 57d9100874be4642e13f904f15d5fa870591b8d4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/741/pull-knative-sandbox-eventing-kafka-upgrade-tests/1408544833194168320) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/741). [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 -->
travis-minke-sap: /test pull-knative-sandbox-eventing-kafka-upgrade-tests
eric-sap: /lgtm

Patch:
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go b/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go
index 08754a7d8..b29010e24 100644
--- a/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go
@@ -25,7 +25,7 @@ import (
 
 var cs = apis.NewBatchConditionSet(
 	ResetOffsetConditionRefMapped,
-	ResetOffsetConditionResetInitiated,
+	ResetOffsetConditionAcquireDataPlaneServices,
 	ResetOffsetConditionConsumerGroupsStopped,
 	ResetOffsetConditionOffsetsUpdated,
 	ResetOffsetConditionConsumerGroupsStarted)
@@ -41,10 +41,9 @@ const (
 	// values will then be populated in the the ResetOffsetStatus.
 	ResetOffsetConditionRefMapped apis.ConditionType = "RefMapped"
 
-	// ResetOffsetConditionResetInitiated has status True when the ResetOffset is being processed
-	// Until that time it is either "Unknown" when first encountered, or "Blocked" if another
-	// ResetOffset is being processed for the same "Ref" resource.
-	ResetOffsetConditionResetInitiated apis.ConditionType = "ResetInitiated"
+	// ResetOffsetConditionAcquireDataPlaneServices has status True if the most recent attempt to
+	// reconcile the control-protocol DataPlane Services from the ConnectionPool was successful.
+	ResetOffsetConditionAcquireDataPlaneServices apis.ConditionType = "AcquireDataPlaneServices"
 
 	// ResetOffsetConditionConsumerGroupsStopped has status True when all of the ConsumerGroups
 	// associated with the referenced object (Subscription, Trigger, etc.) have been stopped.
@@ -85,14 +84,9 @@ func (ros *ResetOffsetStatus) GetCondition(t apis.ConditionType) *apis.Condition
 	return ros.GetConditionSet().Manage(ros).GetCondition(t)
 }
 
-// IsInitiated returns true if the ResetOffsetConditionResetInitiated status is true.
-func (ros *ResetOffsetStatus) IsInitiated() bool {
-	return ros.GetConditionSet().Manage(ros).GetCondition(ResetOffsetConditionResetInitiated).Status == corev1.ConditionTrue
-}
-
-// IsCompleted returns true if the ResetOffsetConditionSucceeded status is not unknown.
-func (ros *ResetOffsetStatus) IsCompleted() bool {
-	return ros.GetConditionSet().Manage(ros).GetCondition(ResetOffsetConditionSucceeded).Status != corev1.ConditionUnknown
+// IsOffsetsUpdated returns true if the ResetOffsetConditionOffsetsUpdates status is true.
+func (ros *ResetOffsetStatus) IsOffsetsUpdated() bool {
+	return ros.GetConditionSet().Manage(ros).GetCondition(ResetOffsetConditionOffsetsUpdated).Status == corev1.ConditionTrue
 }
 
 // IsSucceeded returns true if the ResetOffsetConditionSucceeded status is true.
@@ -113,12 +107,12 @@ func (ros *ResetOffsetStatus) MarkRefMappedTrue() {
 	ros.GetConditionSet().Manage(ros).MarkTrue(ResetOffsetConditionRefMapped)
 }
 
-func (ros *ResetOffsetStatus) MarkResetInitiatedFailed(reason, messageFormat string, messageA ...interface{}) {
-	ros.GetConditionSet().Manage(ros).MarkFalse(ResetOffsetConditionResetInitiated, reason, messageFormat, messageA...)
+func (ros *ResetOffsetStatus) MarkAcquireDataPlaneServicesFailed(reason, messageFormat string, messageA ...interface{}) {
+	ros.GetConditionSet().Manage(ros).MarkFalse(ResetOffsetConditionAcquireDataPlaneServices, reason, messageFormat, messageA...)
 }
 
-func (ros *ResetOffsetStatus) MarkResetInitiatedTrue() {
-	ros.GetConditionSet().Manage(ros).MarkTrue(ResetOffsetConditionResetInitiated)
+func (ros *ResetOffsetStatus) MarkAcquireDataPlaneServicesTrue() {
+	ros.GetConditionSet().Manage(ros).MarkTrue(ResetOffsetConditionAcquireDataPlaneServices)
 }
 
 func (ros *ResetOffsetStatus) MarkConsumerGroupsStoppedFailed(reason, messageFormat string, messageA ...interface{}) {
diff --git a/pkg/common/commands/resetoffset/controller/controller.go b/pkg/common/commands/resetoffset/controller/controller.go
index 8c0b1c355..aaa5b73a2 100644
--- a/pkg/common/commands/resetoffset/controller/controller.go
+++ b/pkg/common/commands/resetoffset/controller/controller.go
@@ -19,6 +19,7 @@ package controller
 import (
 	"context"
 
+	"github.com/google/uuid"
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/types"
 	ctrlreconciler "knative.dev/control-protocol/pkg/reconciler"
@@ -61,6 +62,7 @@ func NewControllerFactory(
 
 		// Create A ResetOffset Reconciler
 		reconciler := &Reconciler{
+			uid:                           types.UID(uuid.NewString()),
 			podLister:                     podInformer.Lister(),
 			resetoffsetLister:             resetoffsetInformer.Lister(),
 			refMapper:                     refMapper,
diff --git a/pkg/common/commands/resetoffset/controller/dataplane.go b/pkg/common/commands/resetoffset/controller/dataplane.go
index ad42ee008..33dbaaf4a 100644
--- a/pkg/common/commands/resetoffset/controller/dataplane.go
+++ b/pkg/common/commands/resetoffset/controller/dataplane.go
@@ -19,7 +19,6 @@ package controller
 import (
 	"context"
 	"fmt"
-	"hash/fnv"
 	"strings"
 	"sync"
 	"time"
@@ -36,14 +35,18 @@ import (
 
 	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
 	"knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers"
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol"
 	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
 )
 
 const (
 	asyncCommandResultPollDuration    = 1 * time.Second  // AsyncCommandResult Polling Duration
 	asyncCommandResultTimeoutDuration = 10 * time.Second // AsyncCommandResult Timeout Duration
+)
 
-	ControlProtocolServerPort = 8085 // TODO - Use common constant once eric delivers!
+var (
+	// TODO - How long should a Successful ResetOffset Reconciliation Take ???
+	asyncCommandLockTimeout = 1 * time.Minute // AsyncCommand ConsumerGroup Lock Timeout (Enough for successful ResetOffset reconciliation)
 )
 
 // reconcileDataPlaneServices updates the Reconciler ConnectionPool Services associated with the specified RefInfo.
@@ -63,7 +66,7 @@ func (r *Reconciler) reconcileDataPlaneServices(ctx context.Context, resetOffset
 	// Append The Control-Protocol Server Port Number To The PodIPs If Not Already Present
 	for index, podIP := range podIPs {
 		if !strings.Contains(podIP, ":") {
-			podIPs[index] = fmt.Sprintf("%s:%d", podIP, ControlProtocolServerPort)
+			podIPs[index] = fmt.Sprintf("%s:%d", podIP, controlprotocol.ServerPort)
 		}
 	}
 	logger.Debug("Detected DataPlane Services", zap.Any("Pod IPs", podIPs))
@@ -154,18 +157,35 @@ func (r *Reconciler) sendConsumerGroupAsyncCommand(ctx context.Context,
 	logger := logging.FromContext(ctx).Desugar().With(zap.String("PodIP", podIP), zap.Int("OpCode", int(opCode)))
 
 	// Generate A CommandID For The ResetOffset
-	commandId, err := generateCommandId(resetOffset, podIP, opCode)
+	commandId, err := GenerateCommandId(resetOffset, podIP, opCode)
 	if err != nil {
 		logger.Error("Failed to generate Command ID for ResetOffset", zap.Error(err))
 		return fmt.Errorf("failed to generate Command ID for ResetOffset: %v", err)
 	}
 
-	// Create And Send A The ConsumerGroupAsyncCommand to The Control-Protocol Service
-	consumerGroupAsyncCommand := commands.NewConsumerGroupAsyncCommand(commandId, refInfo.TopicName, refInfo.GroupId)
+	// Generate A Lock Token For The Current Reconciler & ResetOffset
+	lockToken := GenerateLockToken(r.uid, resetOffset.UID)
+
+	// Create The OpCode Specific CommandLock (Lock Before Stop, Unlock After Start)
+	var commandLock *commands.CommandLock
+	switch opCode {
+	case commands.StopConsumerGroupOpCode:
+		commandLock = commands.NewCommandLock(lockToken, asyncCommandLockTimeout, true, false)
+	case commands.StartConsumerGroupOpCode:
+		commandLock = commands.NewCommandLock(lockToken, 0, false, true)
+	default:
+		logger.Error("Received invalid ConsumerGroupAsyncCommand OpCode", zap.Uint8("OpCode", uint8(opCode)))
+		return fmt.Errorf("received invalid ConsumerGroupAsyncCommand OpCode: %d", uint8(opCode))
+	}
+
+	// Create The ConsumerGroupAsyncCommand With CommandLock
+	consumerGroupAsyncCommand := commands.NewConsumerGroupAsyncCommand(commandId, refInfo.TopicName, refInfo.GroupId, commandLock)
+
+	// Send The ConsumerGroupAsyncCommand & Wait For Acknowledgement
 	err = service.SendAndWaitForAck(opCode, consumerGroupAsyncCommand)
 	if err != nil {
 		logger.Error("Failed to send ConsumerGroup AsyncCommand", zap.Int64("CommandID", commandId), zap.Error(err))
-		return fmt.Errorf("failed to send ConsumerGroup AsyncCommand '%d' : %v", commandId, err)
+		return fmt.Errorf("failed to send ConsumerGroup AsyncCommand '%d': %v", commandId, err)
 	}
 
 	// Wait For The AsyncCommand Result & Return Results
@@ -201,13 +221,3 @@ func (r *Reconciler) waitForAsyncCommandResult(resetOffset *kafkav1alpha1.ResetO
 	// Return The Result
 	return err
 }
-
-// generateCommandId returns an int64 hash based on the specified ResetOffset.
-func generateCommandId(resetOffset *kafkav1alpha1.ResetOffset, podIP string, opCode ctrl.OpCode) (int64, error) {
-	hash := fnv.New32a()
-	_, err := hash.Write([]byte(fmt.Sprintf("%s-%d-%s-%d", string(resetOffset.UID), resetOffset.Generation, podIP, opCode)))
-	if err != nil {
-		return -1, err
-	}
-	return int64(hash.Sum32()), nil
-}
diff --git a/pkg/common/commands/resetoffset/controller/reconciler.go b/pkg/common/commands/resetoffset/controller/reconciler.go
index 1806e4dd7..d9bc6c02d 100644
--- a/pkg/common/commands/resetoffset/controller/reconciler.go
+++ b/pkg/common/commands/resetoffset/controller/reconciler.go
@@ -48,6 +48,7 @@ var (
 
 // Reconciler Implements controller.Reconciler for ResetOffset Resources
 type Reconciler struct {
+	uid                           types.UID
 	kafkaBrokers                  []string
 	saramaConfig                  *sarama.Config
 	podLister                     corev1listers.PodLister
@@ -65,16 +66,16 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, resetOffset *kafkav1alph
 	logger.Debug("<==========  START RESET-OFFSET RECONCILIATION  ==========>")
 
 	//
-	// Ignore Previously Initiated ResetOffset Instances
+	// Ignore Previously Successful ResetOffset Instances
 	//
 	// Normally this would be achieved via a cache.FilteringResourceEventHandler{} directly
 	// in the controller, but that only has a metav1.Object to deal with which does not
 	// expose the ResetOffset.Status we need to make the filtering decision, so we are left
 	// to handle it here in the Reconciler.
 	//
-	if resetOffset.Status.IsInitiated() || resetOffset.Status.IsCompleted() {
-		logger.Debug("Skipping reconciliation of previously executed ResetOffset instance")
-		return reconciler.NewEvent(corev1.EventTypeNormal, ResetOffsetSkipped.String(), "Skipped previously executed ResetOffset")
+	if resetOffset.Status.IsSucceeded() {
+		logger.Debug("Skipping reconciliation of previously successful ResetOffset instance")
+		return reconciler.NewEvent(corev1.EventTypeNormal, ResetOffsetSkipped.String(), "Skipped previously successful ResetOffset")
 	}
 
 	// Reset The ResetOffset's Status Conditions To Unknown
@@ -92,59 +93,59 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, resetOffset *kafkav1alph
 	resetOffset.Status.SetGroup(refInfo.GroupId)
 	resetOffset.Status.MarkRefMappedTrue()
 
-	// Parse The Sarama Offset Time From ResetOffset Spec
-	offsetTime, err := resetOffset.Spec.ParseSaramaOffsetTime()
-	if err != nil {
-		logger.Error("Failed to parse Sarama Offset Time from ResetOffset Spec", zap.Error(err))
-		return err // Should never happen assuming Validation is in place
-	}
-	logger.Info("Successfully parsed Sarama Offset time from ResetOffset Spec", zap.Int64("Time (millis)", offsetTime))
-
 	// Reconcile The DataPlane "Services" From The ConnectionPool For Specified Key
 	dataPlaneServices, err := r.reconcileDataPlaneServices(ctx, resetOffset, refInfo)
 	if err != nil {
 		logger.Error("Failed to reconcile DataPlane services from ConnectionPool", zap.Error(err))
-		// TODO - how to handle this - need another status condition and mark as failed?
+		resetOffset.Status.MarkAcquireDataPlaneServicesFailed("FailedToAcquireDataPlaneServices", "Failed to reconciler DataPlane Services from ConnectionPool: %v", err)
 		return fmt.Errorf("failed to reconcile DataPlane Services from ConnectionPool: %v", err)
 	}
-	logger.Info("Successfully reconciled DataPlane Services", zap.Any("Services", dataPlaneServiceIPs(dataPlaneServices)))
-
-	// TODO - Send "Initiate" Message to Dispatcher Replicas to lock mutext and start process.
-	resetOffset.Status.MarkResetInitiatedTrue()
-
-	// Stop The ConsumerGroup In Associated Dispatchers
-	err = r.stopConsumerGroups(ctx, resetOffset, dataPlaneServices, refInfo)
-	if err != nil {
-		logger.Error("Failed to stop ConsumerGroups", zap.Error(err))
-		resetOffset.Status.MarkConsumerGroupsStoppedFailed("FailedToStopConsumerGroups", "Failed to stop one or more ConsumerGroups: %v", err)
-		// TODO - should we re-reconcile and keep trying (stuck) or rollback here or in stopConsumerGroups() ???
-		return nil
-	}
-	logger.Info("Successfully stopped ConsumerGroups")
-	resetOffset.Status.MarkConsumerGroupsStoppedTrue()
-
-	// Update The Sarama Offsets & Update ResetOffset CRD With OffsetMappings
-	offsetMappings, err := r.reconcileOffsets(ctx, refInfo, offsetTime)
-	if err != nil {
-		logger.Error("Failed to update Offsets of one or more partitions", zap.Error(err))
-		resetOffset.Status.MarkOffsetsUpdatedFailed("FailedToUpdateOffsets", "Failed to update Offsets of one or more Partitions: %v", err)
-		return fmt.Errorf("failed to update offsets of one or more partitions: %v", err)
-	}
-	if offsetMappings != nil {
-		resetOffset.Status.SetPartitions(offsetMappings)
+	logger.Info("Successfully reconciled DataPlane Services from ConnectionPool", zap.Any("Services", dataPlaneServiceIPs(dataPlaneServices)))
+	resetOffset.Status.MarkAcquireDataPlaneServicesTrue()
+
+	// Only Stop ConsumerGroups & Update Offsets Once
+	if !resetOffset.Status.IsOffsetsUpdated() {
+
+		// Parse The Sarama Offset Time From ResetOffset Spec
+		offsetTime, err := resetOffset.Spec.ParseSaramaOffsetTime()
+		if err != nil {
+			logger.Error("Failed to parse Sarama Offset Time from ResetOffset Spec", zap.Error(err))
+			return err // Should never happen assuming Validation is in place
+		}
+		logger.Info("Successfully parsed Sarama Offset Time from ResetOffset Spec", zap.Int64("Time (millis)", offsetTime))
+
+		// Stop The ConsumerGroup In Associated Dispatchers
+		err = r.stopConsumerGroups(ctx, resetOffset, dataPlaneServices, refInfo)
+		if err != nil {
+			logger.Error("Failed to stop one or more ConsumerGroups", zap.Error(err))
+			resetOffset.Status.MarkConsumerGroupsStoppedFailed("FailedToStopConsumerGroups", "Failed to stop one or more ConsumerGroups: %v", err)
+			return fmt.Errorf("failed to stop one or more ConsumerGroups: %v", err)
+		}
+		logger.Info("Successfully stopped all ConsumerGroups")
+		resetOffset.Status.MarkConsumerGroupsStoppedTrue()
+
+		// Update The Sarama Offsets & Update ResetOffset CRD With OffsetMappings (Single Atomic Operation For All Offsets)
+		offsetMappings, err := r.reconcileOffsets(ctx, refInfo, offsetTime)
+		if err != nil {
+			logger.Error("Failed to update Offsets of ConsumerGroup Partitions", zap.Error(err))
+			resetOffset.Status.MarkOffsetsUpdatedFailed("FailedToUpdateOffsets", "Failed to update Offsets of ConsumerGroup Partitions: %v", err)
+			return fmt.Errorf("failed to update Offsets of ConsumerGroup Partitions: %v", err)
+		}
+		if offsetMappings != nil {
+			resetOffset.Status.SetPartitions(offsetMappings)
+		}
+		logger.Info("Successfully updated Offsets of all partitions")
+		resetOffset.Status.MarkOffsetsUpdatedTrue()
 	}
-	logger.Info("Successfully updated Offsets of all partitions")
-	resetOffset.Status.MarkOffsetsUpdatedTrue()
 
 	// Start The ConsumerGroup In Associated Dispatchers
 	err = r.startConsumerGroups(ctx, resetOffset, dataPlaneServices, refInfo)
 	if err != nil {
-		logger.Error("Failed to start ConsumerGroups", zap.Error(err))
-		resetOffset.Status.MarkConsumerGroupsStartedFailed("FailedToStartConsumerGroups", "Failed to start one or more ConsumerGroups: %v", err)
-		// TODO - should we re-reconcile and keep trying (stuck) or rollback here or in stopConsumerGroups() ???
-		return nil // TODO - for now stopping without re-queueing
+		logger.Error("Failed to restart one or more ConsumerGroups", zap.Error(err))
+		resetOffset.Status.MarkConsumerGroupsStartedFailed("FailedToStartConsumerGroups", "Failed to restart one or more ConsumerGroups: %v", err)
+		return fmt.Errorf("failed to restart one or more ConsumerGroups: %v", err)
 	}
-	logger.Info("Successfully started ConsumerGroups")
+	logger.Info("Successfully started all ConsumerGroups")
 	resetOffset.Status.MarkConsumerGroupsStartedTrue()
 
 	// Return Reconciled Success Event
@@ -158,12 +159,6 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, resetOffset *kafkav1alpha
 	logger := logging.FromContext(ctx)
 	logger.Debug("<==========  START RESET-OFFSET FINALIZATION  ==========>")
 
-	// Do NOT Finalize ResetOffset Instances That Are "Executing"
-	if resetOffset.Status.IsInitiated() && !resetOffset.Status.IsCompleted() {
-		logger.Debug("Skipping finalization of in-progress ResetOffset instance")
-		return fmt.Errorf("skipping finalization of in-progress ResetOffset instance")
-	}
-
 	// Clean The AsyncCommandNotificationStore
 	r.asyncCommandNotificationStore.CleanPodsNotifications(types.NamespacedName{
 		Namespace: resetOffset.Namespace,
diff --git a/pkg/common/commands/resetoffset/controller/util.go b/pkg/common/commands/resetoffset/controller/util.go
new file mode 100644
index 000000000..a24d8f1d3
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/util.go
@@ -0,0 +1,42 @@
+/*
+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 controller
+
+import (
+	"fmt"
+	"hash/fnv"
+
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "knative.dev/control-protocol/pkg"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+)
+
+// GenerateLockToken returns a string representing a Lock.Token unique to the specified Reconciler and ResetOffset
+func GenerateLockToken(reconcilerId types.UID, resetOffsetId types.UID) string {
+	return fmt.Sprintf("%s-%s", string(reconcilerId), string(resetOffsetId))
+}
+
+// GenerateCommandId returns an int64 hash based on the specified ResetOffset.
+func GenerateCommandId(resetOffset *kafkav1alpha1.ResetOffset, podIP string, opCode ctrl.OpCode) (int64, error) {
+	hash := fnv.New32a()
+	_, err := hash.Write([]byte(fmt.Sprintf("%s-%d-%s-%d", string(resetOffset.UID), resetOffset.Generation, podIP, opCode)))
+	if err != nil {
+		return -1, err
+	}
+	return int64(hash.Sum32()), nil
+}
diff --git a/pkg/common/controlprotocol/commands/consumergroup.go b/pkg/common/controlprotocol/commands/consumergroup.go
index 96fd7e0da..ef31c13a9 100644
--- a/pkg/common/controlprotocol/commands/consumergroup.go
+++ b/pkg/common/controlprotocol/commands/consumergroup.go
@@ -37,19 +37,22 @@ var _ ctrlmessage.AsyncCommand = (*ConsumerGroupAsyncCommand)(nil)
 
 // ConsumerGroupAsyncCommand implements an AsyncCommand for handling ConsumerGroup related operations.
 type ConsumerGroupAsyncCommand struct {
-	Version   int16  `json:"version"`
-	CommandId int64  `json:"id"`
-	TopicName string `json:"topic"`
-	GroupId   string `json:"group"`
+	Version   int16        `json:"version"`
+	CommandId int64        `json:"commandId"`
+	TopicName string       `json:"topicName"`
+	GroupId   string       `json:"groupId"`
+	Lock      *CommandLock `json:"lock,omitempty"`
 }
 
 // NewConsumerGroupAsyncCommand constructs and returns a new ConsumerGroupAsyncCommand.
-func NewConsumerGroupAsyncCommand(commandId int64, topicName string, groupId string) *ConsumerGroupAsyncCommand {
+func NewConsumerGroupAsyncCommand(commandId int64, topicName string, groupId string, lock *CommandLock) *ConsumerGroupAsyncCommand {
+
 	return &ConsumerGroupAsyncCommand{
 		Version:   ConsumerGroupAsyncCommandVersion, // Only One Version For Now - Validate Compatibility In Handler ; )
 		CommandId: commandId,
 		TopicName: topicName,
 		GroupId:   groupId,
+		Lock:      lock,
 	}
 }
 
diff --git a/pkg/common/controlprotocol/commands/lock.go b/pkg/common/controlprotocol/commands/lock.go
new file mode 100644
index 000000000..8a8a2723a
--- /dev/null
+++ b/pkg/common/controlprotocol/commands/lock.go
@@ -0,0 +1,37 @@
+/*
+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 commands
+
+import "time"
+
+// CommandLock provides common meta-data for adding/removing locks before/after commands.
+type CommandLock struct {
+	Token       string        `json:"token"`       // A Token Identifying The Owner Of The Lock
+	Timeout     time.Duration `json:"timeout"`     // Custom Lock Timeout, If 0 Then Default Timeout May Apply
+	LockBefore  bool          `json:"lockBefore"`  // Optionally Lock The Resource Before The Command
+	UnlockAfter bool          `json:"unlockAfter"` // Optionally Unlock The Resource After The Command
+}
+
+// NewCommandLock is a convenience constructor for the CommandLock struct.
+func NewCommandLock(token string, timeout time.Duration, lockBefore bool, unlockAfter bool) *CommandLock {
+	return &CommandLock{
+		Token:       token,
+		Timeout:     timeout,
+		LockBefore:  lockBefore,
+		UnlockAfter: unlockAfter,
+	}
+}


Test patch:
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle_test.go b/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle_test.go
index a02d9e8e1..3cd3a37ac 100644
--- a/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle_test.go
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle_test.go
@@ -28,12 +28,12 @@ import (
 )
 
 var (
-	condSucceeded             = apis.Condition{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionTrue}
-	condRefMapped             = apis.Condition{Type: ResetOffsetConditionRefMapped, Status: corev1.ConditionTrue}
-	condResetInitiated        = apis.Condition{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionTrue}
-	condConsumerGroupsStopped = apis.Condition{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionTrue}
-	condOffsetsUpdated        = apis.Condition{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionTrue}
-	condConsumerGroupsStarted = apis.Condition{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionTrue}
+	condSucceeded                 = apis.Condition{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionTrue}
+	condRefMapped                 = apis.Condition{Type: ResetOffsetConditionRefMapped, Status: corev1.ConditionTrue}
+	condAcquiredDataPlaneServices = apis.Condition{Type: ResetOffsetConditionAcquireDataPlaneServices, Status: corev1.ConditionTrue}
+	condConsumerGroupsStopped     = apis.Condition{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionTrue}
+	condOffsetsUpdated            = apis.Condition{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionTrue}
+	condConsumerGroupsStarted     = apis.Condition{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionTrue}
 )
 
 func TestResetOffset_GetConditionSet(t *testing.T) {
@@ -71,14 +71,14 @@ func TestResetOffsetStatus_GetCondition(t *testing.T) {
 			wantCondition: &condRefMapped,
 		},
 		{
-			name: "ResetOffsetConditionResetInitiated",
+			name: "ResetOffsetConditionAcquireDataPlaneServices",
 			status: &ResetOffsetStatus{
 				Status: duckv1.Status{
-					Conditions: []apis.Condition{condResetInitiated},
+					Conditions: []apis.Condition{condAcquiredDataPlaneServices},
 				},
 			},
-			conditionType: ResetOffsetConditionResetInitiated,
-			wantCondition: &condResetInitiated,
+			conditionType: ResetOffsetConditionAcquireDataPlaneServices,
+			wantCondition: &condAcquiredDataPlaneServices,
 		},
 		{
 			name: "ResetOffsetConditionConsumerGroupsStopped",
@@ -134,11 +134,11 @@ func TestResetOffsetStatus_InitializeConditions(t *testing.T) {
 			wantStatus: &ResetOffsetStatus{
 				Status: duckv1.Status{
 					Conditions: []apis.Condition{
+						{Type: ResetOffsetConditionAcquireDataPlaneServices, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionRefMapped, Status: corev1.ConditionUnknown},
-						{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionUnknown},
 					},
 				},
@@ -156,11 +156,11 @@ func TestResetOffsetStatus_InitializeConditions(t *testing.T) {
 			wantStatus: &ResetOffsetStatus{
 				Status: duckv1.Status{
 					Conditions: []apis.Condition{
+						{Type: ResetOffsetConditionAcquireDataPlaneServices, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionFalse},
 						{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionRefMapped, Status: corev1.ConditionUnknown},
-						{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionUnknown},
 					},
 				},
@@ -178,11 +178,11 @@ func TestResetOffsetStatus_InitializeConditions(t *testing.T) {
 			wantStatus: &ResetOffsetStatus{
 				Status: duckv1.Status{
 					Conditions: []apis.Condition{
+						{Type: ResetOffsetConditionAcquireDataPlaneServices, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionTrue},
 						{Type: ResetOffsetConditionRefMapped, Status: corev1.ConditionUnknown},
-						{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionUnknown},
 					},
 				},
@@ -200,11 +200,11 @@ func TestResetOffsetStatus_InitializeConditions(t *testing.T) {
 			wantStatus: &ResetOffsetStatus{
 				Status: duckv1.Status{
 					Conditions: []apis.Condition{
+						{Type: ResetOffsetConditionAcquireDataPlaneServices, Status: corev1.ConditionTrue},
 						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionTrue},
 						{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionTrue},
 						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionTrue},
 						{Type: ResetOffsetConditionRefMapped, Status: corev1.ConditionTrue},
-						{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionTrue},
 						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionTrue},
 					},
 				},
@@ -224,181 +224,133 @@ func TestResetOffsetStatus_InitializeConditions(t *testing.T) {
 	}
 }
 
-func TestResetOffsetStatus_IsInitiated(t *testing.T) {
+func TestResetOffsetStatus_IsOffsetsUpdated(t *testing.T) {
 
 	tests := []struct {
-		name          string
-		markInitiated corev1.ConditionStatus
-		wantInitiated bool
+		name        string
+		markUpdated corev1.ConditionStatus
+		wantUpdated bool
 	}{
 		{
-			name:          "Initiated True",
-			markInitiated: corev1.ConditionTrue,
-			wantInitiated: true,
+			name:        "Offsets Updated True",
+			markUpdated: corev1.ConditionTrue,
+			wantUpdated: true,
 		},
 		{
-			name:          "Initiated False",
-			markInitiated: corev1.ConditionFalse,
-			wantInitiated: false,
+			name:        "Offsets Updated False",
+			markUpdated: corev1.ConditionFalse,
+			wantUpdated: false,
 		},
 		{
-			name:          "Initiated Unknown",
-			markInitiated: corev1.ConditionUnknown,
-			wantInitiated: false,
+			name:        "Offsets Updated Unknown",
+			markUpdated: corev1.ConditionUnknown,
+			wantUpdated: false,
 		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			resetOffsetStatus := &ResetOffsetStatus{}
-			resetOffsetStatus.InitializeConditions()
 
-			switch test.markInitiated {
-			case corev1.ConditionTrue:
-				resetOffsetStatus.MarkResetInitiatedTrue()
-			case corev1.ConditionFalse:
-				resetOffsetStatus.MarkResetInitiatedFailed("TestingOffsetsResetInitiatedStatus", "TestMessage")
-			case corev1.ConditionUnknown:
-				for _, condition := range resetOffsetStatus.Conditions {
-					assert.True(t, condition.IsUnknown()) // Verify Initiated Is "Unknown" From InitializeConditions()
-				}
-			}
-			assert.Equal(t, test.wantInitiated, resetOffsetStatus.IsInitiated())
-		})
-	}
-}
-
-func TestResetOffsetStatus_IsCompleted(t *testing.T) {
-
-	tests := []struct {
-		name          string
-		markSucceeded corev1.ConditionStatus
-		wantCompleted bool
-	}{
-		{
-			name:          "Succeeded True",
-			markSucceeded: corev1.ConditionTrue,
-			wantCompleted: true,
-		},
-		{
-			name:          "Succeeded False",
-			markSucceeded: corev1.ConditionFalse,
-			wantCompleted: true,
-		},
-		{
-			name:          "Succeeded Unknown",
-			markSucceeded: corev1.ConditionUnknown,
-			wantCompleted: false,
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
 			resetOffsetStatus := &ResetOffsetStatus{}
 			resetOffsetStatus.InitializeConditions()
 
-			switch test.markSucceeded {
+			switch test.markUpdated {
 			case corev1.ConditionTrue:
-				resetOffsetStatus.MarkRefMappedTrue()
-				resetOffsetStatus.MarkResetInitiatedTrue()
-				resetOffsetStatus.MarkConsumerGroupsStoppedTrue()
 				resetOffsetStatus.MarkOffsetsUpdatedTrue()
-				resetOffsetStatus.MarkConsumerGroupsStartedTrue()
 			case corev1.ConditionFalse:
-				resetOffsetStatus.MarkRefMappedFailed("TestingOffsetsRefMappedStatus", "TestMessage")
-				resetOffsetStatus.MarkResetInitiatedFailed("TestingOffsetsResetInitiatedStatus", "TestMessage")
-				resetOffsetStatus.MarkConsumerGroupsStoppedFailed("TestingConsumerGroupsStoppedStatus", "TestMessage")
 				resetOffsetStatus.MarkOffsetsUpdatedFailed("TestingOffsetsUpdatedFailedStatus", "TestMessage")
-				resetOffsetStatus.MarkConsumerGroupsStartedFailed("TestingConsumerGroupsStartedStatus", "TestMessage")
 			case corev1.ConditionUnknown:
 				for _, condition := range resetOffsetStatus.Conditions {
-					assert.True(t, condition.IsUnknown()) // Verify Succeeded Is "Unknown" From InitializeConditions()
+					assert.True(t, condition.IsUnknown()) // Verify OffsetsUpdated Is "Unknown" From InitializeConditions()
 				}
 			}
-			assert.Equal(t, test.wantCompleted, resetOffsetStatus.IsCompleted())
+
+			assert.Equal(t, test.wantUpdated, resetOffsetStatus.IsOffsetsUpdated())
 		})
 	}
 }
 
 func TestResetOffsetStatus_IsSucceeded(t *testing.T) {
 	tests := []struct {
-		name                      string
-		markRefMapped             bool
-		markResetInitiated        bool
-		markConsumerGroupsStopped bool
-		markOffsetsUpdated        bool
-		markConsumerGroupsStarted bool
-		wantSucceeded             bool
+		name                         string
+		markRefMapped                bool
+		markAcquireDataPlaneServices bool
+		markConsumerGroupsStopped    bool
+		markOffsetsUpdated           bool
+		markConsumerGroupsStarted    bool
+		wantSucceeded                bool
 	}{
 		{
-			name:                      "Happy",
-			markRefMapped:             true,
-			markResetInitiated:        true,
-			markConsumerGroupsStopped: true,
-			markOffsetsUpdated:        true,
-			markConsumerGroupsStarted: true,
-			wantSucceeded:             true,
+			name:                         "Happy",
+			markRefMapped:                true,
+			markAcquireDataPlaneServices: true,
+			markConsumerGroupsStopped:    true,
+			markOffsetsUpdated:           true,
+			markConsumerGroupsStarted:    true,
+			wantSucceeded:                true,
 		},
 		{
-			name:                      "RefMapped Failed",
-			markRefMapped:             false,
-			markResetInitiated:        true,
-			markConsumerGroupsStopped: true,
-			markOffsetsUpdated:        true,
-			markConsumerGroupsStarted: true,
-			wantSucceeded:             false,
+			name:                         "RefMapped Failed",
+			markRefMapped:                false,
+			markAcquireDataPlaneServices: true,
+			markConsumerGroupsStopped:    true,
+			markOffsetsUpdated:           true,
+			markConsumerGroupsStarted:    true,
+			wantSucceeded:                false,
 		},
 		{
-			name:                      "ResetInitiated Failed",
-			markRefMapped:             true,
-			markResetInitiated:        false,
-			markConsumerGroupsStopped: true,
-			markOffsetsUpdated:        true,
-			markConsumerGroupsStarted: true,
-			wantSucceeded:             false,
+			name:                         "AcquireDataPlaneServices Failed",
+			markRefMapped:                true,
+			markAcquireDataPlaneServices: false,
+			markConsumerGroupsStopped:    true,
+			markOffsetsUpdated:           true,
+			markConsumerGroupsStarted:    true,
+			wantSucceeded:                false,
 		},
 		{
-			name:                      "ConsumerGroupsStopped Failed",
-			markRefMapped:             true,
-			markResetInitiated:        true,
-			markConsumerGroupsStopped: false,
-			markOffsetsUpdated:        true,
-			markConsumerGroupsStarted: true,
-			wantSucceeded:             false,
+			name:                         "ConsumerGroupsStopped Failed",
+			markRefMapped:                true,
+			markAcquireDataPlaneServices: true,
+			markConsumerGroupsStopped:    false,
+			markOffsetsUpdated:           true,
+			markConsumerGroupsStarted:    true,
+			wantSucceeded:                false,
 		},
 		{
-			name:                      "OffsetsUpdated Failed",
-			markRefMapped:             true,
-			markResetInitiated:        true,
-			markConsumerGroupsStopped: true,
-			markOffsetsUpdated:        false,
-			markConsumerGroupsStarted: true,
-			wantSucceeded:             false,
+			name:                         "OffsetsUpdated Failed",
+			markRefMapped:                true,
+			markAcquireDataPlaneServices: true,
+			markConsumerGroupsStopped:    true,
+			markOffsetsUpdated:           false,
+			markConsumerGroupsStarted:    true,
+			wantSucceeded:                false,
 		},
 		{
-			name:                      "ConsumerGroupsStarted Failed",
-			markRefMapped:             true,
-			markResetInitiated:        true,
-			markConsumerGroupsStopped: true,
-			markOffsetsUpdated:        true,
-			markConsumerGroupsStarted: false,
-			wantSucceeded:             false,
+			name:                         "ConsumerGroupsStarted Failed",
+			markRefMapped:                true,
+			markAcquireDataPlaneServices: true,
+			markConsumerGroupsStopped:    true,
+			markOffsetsUpdated:           true,
+			markConsumerGroupsStarted:    false,
+			wantSucceeded:                false,
 		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
+
 			resetOffsetStatus := &ResetOffsetStatus{}
 			resetOffsetStatus.InitializeConditions()
+
 			if test.markRefMapped {
 				resetOffsetStatus.MarkRefMappedTrue()
 			} else {
 				resetOffsetStatus.MarkRefMappedFailed("TestingOffsetsRefMappedStatus", "TestMessage")
 			}
-			if test.markResetInitiated {
-				resetOffsetStatus.MarkResetInitiatedTrue()
+			if test.markAcquireDataPlaneServices {
+				resetOffsetStatus.MarkAcquireDataPlaneServicesTrue()
 			} else {
-				resetOffsetStatus.MarkResetInitiatedFailed("TestingOffsetsResetInitiatedStatus", "TestMessage")
+				resetOffsetStatus.MarkAcquireDataPlaneServicesFailed("TestingAcquireDataPlaneServicesStatus", "TestMessage")
 			}
 			if test.markConsumerGroupsStopped {
 				resetOffsetStatus.MarkConsumerGroupsStoppedTrue()
@@ -415,6 +367,7 @@ func TestResetOffsetStatus_IsSucceeded(t *testing.T) {
 			} else {
 				resetOffsetStatus.MarkConsumerGroupsStartedFailed("TestingConsumerGroupsStartedStatus", "TestMessage")
 			}
+
 			assert.Equal(t, test.wantSucceeded, resetOffsetStatus.IsSucceeded())
 		})
 	}
diff --git a/pkg/common/commands/resetoffset/controller/controller_test.go b/pkg/common/commands/resetoffset/controller/controller_test.go
index 0d3c2a421..618fe037e 100644
--- a/pkg/common/commands/resetoffset/controller/controller_test.go
+++ b/pkg/common/commands/resetoffset/controller/controller_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"k8s.io/client-go/rest"
 	"knative.dev/pkg/client/injection/kube/client/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake" // Knative Fake Informer Injection
 	"knative.dev/pkg/injection"
 	"knative.dev/pkg/injection/sharedmain"
 	"knative.dev/pkg/logging"
@@ -30,14 +31,13 @@ import (
 
 	fakekafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/fake" // Force Fake Informer Injection
-	controllertesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller/testing"
 	refmapperstesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers/testing"
 	configtesting "knative.dev/eventing-kafka/pkg/common/config/testing"
 	"knative.dev/eventing-kafka/pkg/common/configmaploader"
 	fakeConfigmapLoader "knative.dev/eventing-kafka/pkg/common/configmaploader/fake"
 	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
+	controlprotocoltesting "knative.dev/eventing-kafka/pkg/common/controlprotocol/testing"
 	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
-	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake" // Knative Fake Informer Injection
 )
 
 // Test The NewControllerFactory() Functionality
@@ -75,7 +75,7 @@ func TestNewControllerFactory(t *testing.T) {
 	mockResetOffsetRefMapperFactory.On("Create", ctx).Return(mockResetOffsetRefMapper)
 
 	// Create Mock ConnectionPool For Testing
-	mockConnectionPool := &controllertesting.MockConnectionPool{}
+	mockConnectionPool := &controlprotocoltesting.MockConnectionPool{}
 
 	// Verify The ResetOffset ControllerFactory Creates A ControllerConstructor
 	controllerConstructor := NewControllerFactory(mockResetOffsetRefMapperFactory, mockConnectionPool)
diff --git a/pkg/common/commands/resetoffset/controller/dataplane_test.go b/pkg/common/commands/resetoffset/controller/dataplane_test.go
index db59ec080..2246f1e87 100644
--- a/pkg/common/commands/resetoffset/controller/dataplane_test.go
+++ b/pkg/common/commands/resetoffset/controller/dataplane_test.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"go.uber.org/multierr"
@@ -34,7 +35,9 @@ import (
 
 	controllertesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller/testing"
 	refmapperstesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers/testing"
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol"
 	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
+	controlprotocoltesting "knative.dev/eventing-kafka/pkg/common/controlprotocol/testing"
 )
 
 func TestReconciler_ReconcileDataPlaneServices(t *testing.T) {
@@ -42,15 +45,15 @@ func TestReconciler_ReconcileDataPlaneServices(t *testing.T) {
 	// Test Data
 	podIp1 := "1.2.3.4"
 	podIp2 := "2.3.4.5"
-	podIpPort1 := fmt.Sprintf("%s:%d", podIp1, ControlProtocolServerPort)
-	podIpPort2 := fmt.Sprintf("%s:%d", podIp2, ControlProtocolServerPort)
+	podIpPort1 := fmt.Sprintf("%s:%d", podIp1, controlprotocol.ServerPort)
+	podIpPort2 := fmt.Sprintf("%s:%d", podIp2, controlprotocol.ServerPort)
 	podIpPorts := []string{podIpPort1, podIpPort2}
 	pods := []*corev1.Pod{
 		{Status: corev1.PodStatus{PodIP: podIp1}},
 		{Status: corev1.PodStatus{PodIP: podIp2}},
 	}
-	mockDataPlaneService1 := &controllertesting.MockService{}
-	mockDataPlaneService2 := &controllertesting.MockService{}
+	mockDataPlaneService1 := &controlprotocoltesting.MockService{}
+	mockDataPlaneService2 := &controlprotocoltesting.MockService{}
 	testErr := fmt.Errorf("test-error")
 
 	// Create A Context With Test Logger
@@ -105,7 +108,7 @@ func TestReconciler_ReconcileDataPlaneServices(t *testing.T) {
 			mockPodLister.On("Pods", refInfo.DataPlaneNamespace).Return(mockPodNamespaceLister)
 
 			// Create A Mock Control-Protocol ConnectionPool
-			mockConnectionPool := &controllertesting.MockConnectionPool{}
+			mockConnectionPool := &controlprotocoltesting.MockConnectionPool{}
 			if test.podListerErr == nil {
 				mockConnectionPool.On("ReconcileConnections",
 					ctx,
@@ -117,7 +120,7 @@ func TestReconciler_ReconcileDataPlaneServices(t *testing.T) {
 			}
 
 			// Create A Mock Control-Protocol AsyncCommandNotificationStore
-			mockAsyncCommandNotificationStore := &controllertesting.MockAsyncCommandNotificationStore{}
+			mockAsyncCommandNotificationStore := &controlprotocoltesting.MockAsyncCommandNotificationStore{}
 
 			// Create A Reconciler To Test
 			reconciler := &Reconciler{
@@ -150,6 +153,7 @@ func TestReconciler_StopConsumerGroups(t *testing.T) {
 func performStartStopConsumerGroupAsyncCommandsTest(t *testing.T, opCode ctrl.OpCode) {
 
 	// Test Data
+	reconcilerUID := types.UID(uuid.NewString())
 	podIp1 := "1.2.3.4"
 	podIp2 := "2.3.4.5"
 	testErr := fmt.Errorf("test-error")
@@ -191,31 +195,44 @@ func performStartStopConsumerGroupAsyncCommandsTest(t *testing.T, opCode ctrl.Op
 			refInfo := refmapperstesting.NewRefInfo()
 
 			// Determine The Expected CommandIDs
-			commandId1, err := generateCommandId(resetOffset, podIp1, opCode)
+			commandId1, err := GenerateCommandId(resetOffset, podIp1, opCode)
 			assert.Nil(t, err)
-			commandId2, err := generateCommandId(resetOffset, podIp2, opCode)
+			commandId2, err := GenerateCommandId(resetOffset, podIp2, opCode)
 			assert.Nil(t, err)
 
+			// Determine The Expected CommandLock Based On OpCode
+			lockToken := GenerateLockToken(reconcilerUID, resetOffset.UID)
+			var commandLock *commands.CommandLock
+			switch opCode {
+			case commands.StopConsumerGroupOpCode:
+				commandLock = commands.NewCommandLock(lockToken, asyncCommandLockTimeout, true, false)
+			case commands.StartConsumerGroupOpCode:
+				commandLock = commands.NewCommandLock(lockToken, 0, false, true)
+			default:
+				assert.Fail(t, "Unexpected OpCode - Unable To Determine Expected CommandLock")
+			}
+
 			// Create Test ConsumerGroupAsyncCommands For Each Pod
-			consumerGroupAsyncCommand1 := &commands.ConsumerGroupAsyncCommand{Version: 1, CommandId: commandId1, TopicName: refInfo.TopicName, GroupId: refInfo.GroupId}
-			consumerGroupAsyncCommand2 := &commands.ConsumerGroupAsyncCommand{Version: 1, CommandId: commandId2, TopicName: refInfo.TopicName, GroupId: refInfo.GroupId}
+			consumerGroupAsyncCommand1 := &commands.ConsumerGroupAsyncCommand{Version: 1, CommandId: commandId1, TopicName: refInfo.TopicName, GroupId: refInfo.GroupId, Lock: commandLock}
+			consumerGroupAsyncCommand2 := &commands.ConsumerGroupAsyncCommand{Version: 1, CommandId: commandId2, TopicName: refInfo.TopicName, GroupId: refInfo.GroupId, Lock: commandLock}
 
 			// Create A Mock Control-Protocol AsyncCommandNotificationStore & Assign To Reconciler
-			mockAsyncCommandNotificationStore := &controllertesting.MockAsyncCommandNotificationStore{}
+			mockAsyncCommandNotificationStore := &controlprotocoltesting.MockAsyncCommandNotificationStore{}
 			if test.result != nil {
 				mockAsyncCommandNotificationStore.On("GetCommandResult", resetOffsetNamespacedName, podIp1, consumerGroupAsyncCommand1).Return(test.result)
 				mockAsyncCommandNotificationStore.On("GetCommandResult", resetOffsetNamespacedName, podIp2, consumerGroupAsyncCommand2).Return(test.result)
 			}
 
 			// Create The Mock Services To Test Against
-			mockDataPlaneService1 := &controllertesting.MockService{}
-			mockDataPlaneService2 := &controllertesting.MockService{}
+			mockDataPlaneService1 := &controlprotocoltesting.MockService{}
+			mockDataPlaneService2 := &controlprotocoltesting.MockService{}
 			mockDataPlaneService1.On("SendAndWaitForAck", opCode, consumerGroupAsyncCommand1).Return(test.sendErr)
 			mockDataPlaneService2.On("SendAndWaitForAck", opCode, consumerGroupAsyncCommand2).Return(test.sendErr)
 			services := map[string]ctrl.Service{podIp1: mockDataPlaneService1, podIp2: mockDataPlaneService2}
 
 			// Create A Reconciler To Test
 			reconciler := &Reconciler{
+				uid:                           reconcilerUID,
 				asyncCommandNotificationStore: mockAsyncCommandNotificationStore,
 			}
 
@@ -227,15 +244,15 @@ func performStartStopConsumerGroupAsyncCommandsTest(t *testing.T, opCode ctrl.Op
 			case commands.StopConsumerGroupOpCode:
 				actualErr = reconciler.stopConsumerGroups(ctx, resetOffset, services, refInfo)
 			default:
-				assert.Fail(t, "Unexpected OpCode")
+				assert.Fail(t, "Unexpected OpCode - Unable To Determine Test Operation")
 			}
 
 			// Setup Expected Errors Based On Test
 			var expectedErrs []error
 			if test.sendErr != nil {
 				expectedErrs = []error{
-					fmt.Errorf("failed to send ConsumerGroup AsyncCommand '%d' : %v", commandId1, testErr),
-					fmt.Errorf("failed to send ConsumerGroup AsyncCommand '%d' : %v", commandId2, testErr),
+					fmt.Errorf("failed to send ConsumerGroup AsyncCommand '%d': %v", commandId1, testErr),
+					fmt.Errorf("failed to send ConsumerGroup AsyncCommand '%d': %v", commandId2, testErr),
 				}
 			}
 
diff --git a/pkg/common/commands/resetoffset/controller/reconciler_test.go b/pkg/common/commands/resetoffset/controller/reconciler_test.go
index 4f21577d0..d3eb801a5 100644
--- a/pkg/common/commands/resetoffset/controller/reconciler_test.go
+++ b/pkg/common/commands/resetoffset/controller/reconciler_test.go
@@ -23,6 +23,7 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	corev1 "k8s.io/api/core/v1"
@@ -47,7 +48,9 @@ import (
 	controllertesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller/testing"
 	refmapperstesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers/testing"
 	"knative.dev/eventing-kafka/pkg/common/constants"
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol"
 	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
+	controlprotocoltesting "knative.dev/eventing-kafka/pkg/common/controlprotocol/testing"
 	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
 )
 
@@ -57,6 +60,8 @@ func TestReconcile(t *testing.T) {
 	// Test Data
 	logger := logtesting.TestLogger(t)
 
+	reconcilerUID := types.UID(uuid.NewString())
+
 	kafkaBrokers := []string{controllertesting.Brokers}
 	saramaConfig := sarama.NewConfig()
 	topicName := controllertesting.TopicName
@@ -75,7 +80,7 @@ func TestReconcile(t *testing.T) {
 
 	podIp := "1.2.3.4"
 	pods := []*corev1.Pod{{Status: corev1.PodStatus{PodIP: podIp}}}
-	podIpPort := fmt.Sprintf("%s:%d", podIp, ControlProtocolServerPort)
+	podIpPort := fmt.Sprintf("%s:%d", podIp, controlprotocol.ServerPort)
 	podIpPorts := []string{podIpPort}
 
 	testErr := fmt.Errorf("test-error")
@@ -113,7 +118,7 @@ func TestReconcile(t *testing.T) {
 						controllertesting.WithStatusGroup(groupId),
 						controllertesting.WithStatusPartitions(offsetMappings),
 						controllertesting.WithStatusRefMapped(true),
-						controllertesting.WithStatusResetInitiated(true),
+						controllertesting.WithStatusAcquireDataPlaneServices(true),
 						controllertesting.WithStatusConsumerGroupsStopped(true),
 						controllertesting.WithStatusOffsetsUpdated(true),
 						controllertesting.WithStatusConsumerGroupsStarted(true)),
@@ -129,38 +134,19 @@ func TestReconcile(t *testing.T) {
 		//
 
 		{
-			Name: "Skipping Previously Initiated",
-			Key:  controllertesting.ResetOffsetKey,
-			Objects: []runtime.Object{
-				controllertesting.NewResetOffset(controllertesting.WithFinalizer,
-					controllertesting.WithStatusResetInitiated(true)),
-			},
-			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
-				{
-					Object: controllertesting.NewResetOffset(
-						controllertesting.WithFinalizer,
-						controllertesting.WithStatusInitialized,
-						controllertesting.WithStatusResetInitiated(true)),
-				},
-			},
-			WantEvents: []string{
-				Eventf(corev1.EventTypeNormal, ResetOffsetSkipped.String(), "Skipped previously executed ResetOffset"),
-			},
-		},
-		{
-			Name: "Skipping Previously Completed",
+			Name: "Skipping Previously Succeeded",
 			Key:  controllertesting.ResetOffsetKey,
 			Objects: []runtime.Object{
 				controllertesting.NewResetOffset(
 					controllertesting.WithFinalizer,
 					controllertesting.WithStatusRefMapped(true),
-					controllertesting.WithStatusResetInitiated(true),
+					controllertesting.WithStatusAcquireDataPlaneServices(true),
 					controllertesting.WithStatusConsumerGroupsStopped(true),
 					controllertesting.WithStatusOffsetsUpdated(true),
 					controllertesting.WithStatusConsumerGroupsStarted(true)),
 			},
 			WantEvents: []string{
-				Eventf(corev1.EventTypeNormal, ResetOffsetSkipped.String(), "Skipped previously executed ResetOffset"),
+				Eventf(corev1.EventTypeNormal, ResetOffsetSkipped.String(), "Skipped previously successful ResetOffset"),
 			},
 		},
 
@@ -198,7 +184,8 @@ func TestReconcile(t *testing.T) {
 						controllertesting.WithStatusInitialized,
 						controllertesting.WithStatusTopic(topicName),
 						controllertesting.WithStatusGroup(groupId),
-						controllertesting.WithStatusRefMapped(true)),
+						controllertesting.WithStatusRefMapped(true),
+						controllertesting.WithStatusAcquireDataPlaneServices(false, "FailedToAcquireDataPlaneServices", "Failed to reconciler DataPlane Services from ConnectionPool: test-error")),
 				},
 			},
 			WantErr: true,
@@ -219,11 +206,14 @@ func TestReconcile(t *testing.T) {
 						controllertesting.WithStatusTopic(topicName),
 						controllertesting.WithStatusGroup(groupId),
 						controllertesting.WithStatusRefMapped(true),
-						controllertesting.WithStatusResetInitiated(true),
-						controllertesting.WithStatusConsumerGroupsStopped(false, "FailedToStopConsumerGroups", fmt.Sprintf("Failed to stop one or more ConsumerGroups: failed to send ConsumerGroup AsyncCommand '3866008807' : %v", testErr.Error()))),
+						controllertesting.WithStatusAcquireDataPlaneServices(true),
+						controllertesting.WithStatusConsumerGroupsStopped(false, "FailedToStopConsumerGroups", fmt.Sprintf("Failed to stop one or more ConsumerGroups: failed to send ConsumerGroup AsyncCommand '3866008807': %v", testErr.Error()))),
 				},
 			},
-			WantErr: false,
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(corev1.EventTypeWarning, "InternalError", fmt.Sprintf("failed to stop one or more ConsumerGroups: failed to send ConsumerGroup AsyncCommand '3866008807': %v", testErr.Error())),
+			},
 		},
 		{
 			Name:          "Reconcile Offsets Error",
@@ -238,14 +228,14 @@ func TestReconcile(t *testing.T) {
 						controllertesting.WithStatusTopic(topicName),
 						controllertesting.WithStatusGroup(groupId),
 						controllertesting.WithStatusRefMapped(true),
-						controllertesting.WithStatusResetInitiated(true),
+						controllertesting.WithStatusAcquireDataPlaneServices(true),
 						controllertesting.WithStatusConsumerGroupsStopped(true),
-						controllertesting.WithStatusOffsetsUpdated(false, "FailedToUpdateOffsets", "Failed to update Offsets of one or more Partitions: test-error")),
+						controllertesting.WithStatusOffsetsUpdated(false, "FailedToUpdateOffsets", "Failed to update Offsets of ConsumerGroup Partitions: test-error")),
 				},
 			},
 			WantErr: true,
 			WantEvents: []string{
-				Eventf(corev1.EventTypeWarning, "InternalError", fmt.Sprintf("failed to update offsets of one or more partitions: %v", testErr.Error())),
+				Eventf(corev1.EventTypeWarning, "InternalError", fmt.Sprintf("failed to update Offsets of ConsumerGroup Partitions: %v", testErr.Error())),
 			},
 		},
 		{
@@ -262,13 +252,16 @@ func TestReconcile(t *testing.T) {
 						controllertesting.WithStatusGroup(groupId),
 						controllertesting.WithStatusPartitions(offsetMappings),
 						controllertesting.WithStatusRefMapped(true),
-						controllertesting.WithStatusResetInitiated(true),
+						controllertesting.WithStatusAcquireDataPlaneServices(true),
 						controllertesting.WithStatusConsumerGroupsStopped(true),
 						controllertesting.WithStatusOffsetsUpdated(true),
-						controllertesting.WithStatusConsumerGroupsStarted(false, "FailedToStartConsumerGroups", fmt.Sprintf("Failed to start one or more ConsumerGroups: failed to send ConsumerGroup AsyncCommand '3899564045' : %v", testErr))),
+						controllertesting.WithStatusConsumerGroupsStarted(false, "FailedToStartConsumerGroups", fmt.Sprintf("Failed to restart one or more ConsumerGroups: failed to send ConsumerGroup AsyncCommand '3899564045': %v", testErr))),
 				},
 			},
-			WantErr: false,
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(corev1.EventTypeWarning, "InternalError", fmt.Sprintf("failed to restart one or more ConsumerGroups: failed to send ConsumerGroup AsyncCommand '3899564045': %v", testErr.Error())),
+			},
 		},
 
 		//
@@ -285,7 +278,7 @@ func TestReconcile(t *testing.T) {
 				controllertesting.WithStatusGroup(groupId),
 				controllertesting.WithStatusPartitions(offsetMappings),
 				controllertesting.WithStatusRefMapped(true),
-				controllertesting.WithStatusResetInitiated(true),
+				controllertesting.WithStatusAcquireDataPlaneServices(true),
 				controllertesting.WithStatusConsumerGroupsStopped(true),
 				controllertesting.WithStatusOffsetsUpdated(true),
 				controllertesting.WithStatusConsumerGroupsStarted(true))},
@@ -307,23 +300,6 @@ func TestReconcile(t *testing.T) {
 				Eventf(corev1.EventTypeNormal, ResetOffsetFinalized.String(), "Finalized successfully"),
 			},
 		},
-		{
-			Name: "Finalization Executing Instance",
-			Key:  controllertesting.ResetOffsetKey,
-			Objects: []runtime.Object{controllertesting.NewResetOffset(
-				controllertesting.WithFinalizer,
-				controllertesting.WithDeletionTimestamp,
-				controllertesting.WithStatusTopic(topicName),
-				controllertesting.WithStatusGroup(groupId),
-				controllertesting.WithStatusPartitions(offsetMappings),
-				controllertesting.WithStatusRefMapped(true),
-				controllertesting.WithStatusResetInitiated(true),
-				controllertesting.WithStatusConsumerGroupsStopped(true))},
-			WantErr: true,
-			WantEvents: []string{
-				Eventf(corev1.EventTypeWarning, "InternalError", "skipping finalization of in-progress ResetOffset instance"),
-			},
-		},
 	}
 
 	// Restore Sarama Client / OffsetManager Stubs After Test Completion
@@ -379,21 +355,24 @@ func TestReconcile(t *testing.T) {
 		mockPodLister.On("Pods", refInfo.DataPlaneNamespace).Return(mockPodNamespaceLister)
 
 		// Create Test ConsumerGroupAsyncCommands
-		stopCommandId, err := generateCommandId(controllertesting.NewResetOffset(), podIp, commands.StopConsumerGroupOpCode)
+		lockToken := GenerateLockToken(reconcilerUID, "") // TableTest Resources Don't Have A UID And Setting Them Above Doesn't Help
+		stopCommandLock := commands.NewCommandLock(lockToken, asyncCommandLockTimeout, true, false)
+		startCommandLock := commands.NewCommandLock(lockToken, 0, false, true)
+		stopCommandId, err := GenerateCommandId(controllertesting.NewResetOffset(), podIp, commands.StopConsumerGroupOpCode)
 		assert.Nil(t, err)
-		startCommandId, err := generateCommandId(controllertesting.NewResetOffset(), podIp, commands.StartConsumerGroupOpCode)
+		startCommandId, err := GenerateCommandId(controllertesting.NewResetOffset(), podIp, commands.StartConsumerGroupOpCode)
 		assert.Nil(t, err)
-		stopConsumerGroupAsyncCommand := &commands.ConsumerGroupAsyncCommand{Version: 1, CommandId: stopCommandId, TopicName: refInfo.TopicName, GroupId: refInfo.GroupId}
-		startConsumerGroupAsyncCommand := &commands.ConsumerGroupAsyncCommand{Version: 1, CommandId: startCommandId, TopicName: refInfo.TopicName, GroupId: refInfo.GroupId}
+		stopConsumerGroupAsyncCommand := &commands.ConsumerGroupAsyncCommand{Version: 1, CommandId: stopCommandId, TopicName: refInfo.TopicName, GroupId: refInfo.GroupId, Lock: stopCommandLock}
+		startConsumerGroupAsyncCommand := &commands.ConsumerGroupAsyncCommand{Version: 1, CommandId: startCommandId, TopicName: refInfo.TopicName, GroupId: refInfo.GroupId, Lock: startCommandLock}
 
 		// Create The Mock Service To Test Against
-		mockDataPlaneService := &controllertesting.MockService{}
+		mockDataPlaneService := &controlprotocoltesting.MockService{}
 		mockDataPlaneService.On("SendAndWaitForAck", commands.StopConsumerGroupOpCode, stopConsumerGroupAsyncCommand).Return(stopConsumerGroupsErr)
 		mockDataPlaneService.On("SendAndWaitForAck", commands.StartConsumerGroupOpCode, startConsumerGroupAsyncCommand).Return(startConsumerGroupsErr)
 		services := map[string]ctrl.Service{podIp: mockDataPlaneService}
 
 		// Create A Mock Control-Protocol ConnectionPool
-		mockConnectionPool := &controllertesting.MockConnectionPool{}
+		mockConnectionPool := &controlprotocoltesting.MockConnectionPool{}
 		mockConnectionPool.On("ReconcileConnections",
 			mock.Anything,
 			refInfo.ConnectionPoolKey,
@@ -409,7 +388,7 @@ func TestReconcile(t *testing.T) {
 		// Create A Mock Control-Protocol AsyncCommandNotificationStore & Assign To Reconciler
 		resetOffsetNamespacedName := controllertesting.NewResetOffsetNamespacedName()
 		successResult := &ctrlmessage.AsyncCommandResult{}
-		mockAsyncCommandNotificationStore := &controllertesting.MockAsyncCommandNotificationStore{}
+		mockAsyncCommandNotificationStore := &controlprotocoltesting.MockAsyncCommandNotificationStore{}
 		mockAsyncCommandNotificationStore.On("GetCommandResult", resetOffsetNamespacedName, podIp, stopConsumerGroupAsyncCommand).Return(successResult)
 		mockAsyncCommandNotificationStore.On("GetCommandResult", resetOffsetNamespacedName, podIp, startConsumerGroupAsyncCommand).Return(successResult)
 		mockAsyncCommandNotificationStore.On("CleanPodsNotifications", types.NamespacedName{
@@ -434,6 +413,7 @@ func TestReconcile(t *testing.T) {
 
 		// Create The ResetOffset Reconciler Struct
 		r := &Reconciler{
+			uid:                           reconcilerUID,
 			kafkaBrokers:                  kafkaBrokers,
 			saramaConfig:                  saramaConfig,
 			podLister:                     mockPodLister,
diff --git a/pkg/common/commands/resetoffset/controller/testing/data.go b/pkg/common/commands/resetoffset/controller/testing/data.go
index 434a1e1a9..b4b2309ca 100644
--- a/pkg/common/commands/resetoffset/controller/testing/data.go
+++ b/pkg/common/commands/resetoffset/controller/testing/data.go
@@ -137,12 +137,12 @@ func WithStatusRefMapped(state bool, failed ...string) ResetOffsetOption {
 	}
 }
 
-func WithStatusResetInitiated(state bool, failed ...string) ResetOffsetOption {
+func WithStatusAcquireDataPlaneServices(state bool, failed ...string) ResetOffsetOption {
 	return func(resetOffset *kafkav1alpha1.ResetOffset) {
 		if state {
-			resetOffset.Status.MarkResetInitiatedTrue()
+			resetOffset.Status.MarkAcquireDataPlaneServicesTrue()
 		} else {
-			resetOffset.Status.MarkResetInitiatedFailed(failed[0], failed[1])
+			resetOffset.Status.MarkAcquireDataPlaneServicesFailed(failed[0], failed[1])
 		}
 	}
 }
diff --git a/pkg/common/commands/resetoffset/controller/testing/mock.go b/pkg/common/commands/resetoffset/controller/testing/mock.go
index 1eaf95286..379696466 100644
--- a/pkg/common/commands/resetoffset/controller/testing/mock.go
+++ b/pkg/common/commands/resetoffset/controller/testing/mock.go
@@ -17,18 +17,11 @@ limitations under the License.
 package testing
 
 import (
-	"context"
-	"encoding"
-
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/mock"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/types"
 	listerscorev1 "k8s.io/client-go/listers/core/v1"
-	ctrl "knative.dev/control-protocol/pkg"
-	ctrlmessage "knative.dev/control-protocol/pkg/message"
-	ctrlreconciler "knative.dev/control-protocol/pkg/reconciler"
 )
 
 //
@@ -326,106 +319,6 @@ func WithPartitionOffsetManagerMockAsyncClose() MockPartitionOffsetManagerOption
 	}
 }
 
-// TODO - Move these control-protocol mocks to common/controlprotocol/testing/mocks.go once eric merges!
-
-//
-// Mock Control-Protocol ConnectionPool
-//
-
-var _ ctrlreconciler.ControlPlaneConnectionPool = (*MockConnectionPool)(nil)
-
-type MockConnectionPool struct {
-	mock.Mock
-}
-
-func (c *MockConnectionPool) GetConnectedHosts(key string) []string {
-	args := c.Called(key)
-	return args.Get(0).([]string)
-}
-
-func (c *MockConnectionPool) GetServices(key string) map[string]ctrl.Service {
-	args := c.Called(key)
-	return args.Get(0).(map[string]ctrl.Service)
-}
-
-func (c *MockConnectionPool) ResolveControlInterface(key string, host string) (string, ctrl.Service) {
-	args := c.Called(key, host)
-	return args.String(0), args.Get(1).(ctrl.Service)
-}
-
-func (c *MockConnectionPool) RemoveConnection(ctx context.Context, key string, host string) {
-	c.Called(ctx, key, host)
-}
-
-func (c *MockConnectionPool) RemoveAllConnections(ctx context.Context, key string) {
-	c.Called(ctx, key)
-}
-
-func (c *MockConnectionPool) Close(ctx context.Context) {
-	c.Called(ctx)
-}
-
-func (c *MockConnectionPool) ReconcileConnections(ctx context.Context, key string, wantConnections []string, newServiceCb func(string, ctrl.Service), oldServiceCb func(string)) (map[string]ctrl.Service, error) {
-	args := c.Called(ctx, key, wantConnections, newServiceCb, oldServiceCb)
-	return args.Get(0).(map[string]ctrl.Service), args.Error(1)
-}
-
-func (c *MockConnectionPool) DialControlService(ctx context.Context, key string, host string) (string, ctrl.Service, error) {
-	args := c.Called(ctx, key, host)
-	return args.String(0), args.Get(1).(ctrl.Service), args.Error(2)
-}
-
-//
-// Mock Control-Protocol AsyncCommandNotificationStore
-//
-
-var _ ctrlreconciler.AsyncCommandNotificationStore = (*MockAsyncCommandNotificationStore)(nil)
-
-type MockAsyncCommandNotificationStore struct {
-	mock.Mock
-}
-
-func (a *MockAsyncCommandNotificationStore) GetCommandResult(srcName types.NamespacedName, pod string, command ctrlmessage.AsyncCommand) *ctrlmessage.AsyncCommandResult {
-	args := a.Called(srcName, pod, command)
-	return args.Get(0).(*ctrlmessage.AsyncCommandResult)
-}
-
-func (a *MockAsyncCommandNotificationStore) CleanPodsNotifications(srcName types.NamespacedName) {
-	a.Called(srcName)
-}
-
-func (a *MockAsyncCommandNotificationStore) CleanPodNotification(srcName types.NamespacedName, pod string) {
-	a.Called(srcName, pod)
-}
-
-func (a *MockAsyncCommandNotificationStore) MessageHandler(srcName types.NamespacedName, pod string) ctrl.MessageHandler {
-	args := a.Called(srcName, pod)
-	return args.Get(0).(ctrl.MessageHandler)
-}
-
-//
-// Mock Control-Protocol Service
-//
-
-var _ ctrl.Service = (*MockService)(nil)
-
-type MockService struct {
-	mock.Mock
-}
-
-func (s *MockService) SendAndWaitForAck(opcode ctrl.OpCode, payload encoding.BinaryMarshaler) error {
-	args := s.Called(opcode, payload)
-	return args.Error(0)
-}
-
-func (s *MockService) MessageHandler(handler ctrl.MessageHandler) {
-	s.Called(handler)
-}
-
-func (s *MockService) ErrorHandler(handler ctrl.ErrorHandler) {
-	s.Called(handler)
-}
-
 //
 // Mock K8S PodLister
 //
diff --git a/pkg/common/commands/resetoffset/controller/util_test.go b/pkg/common/commands/resetoffset/controller/util_test.go
new file mode 100644
index 000000000..33e6a54df
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/util_test.go
@@ -0,0 +1,66 @@
+/*
+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 controller
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
+)
+
+func TestGenerateLockToken(t *testing.T) {
+
+	// Test Data
+	reconcilerId := types.UID(uuid.NewString())
+	resetOffsetId := types.UID(uuid.NewString())
+
+	// Perform The Test
+	actualLockToken := GenerateLockToken(reconcilerId, resetOffsetId)
+
+	// Verify The Results
+	expectedLockToken := fmt.Sprintf("%s-%s", string(reconcilerId), string(resetOffsetId))
+	assert.Equal(t, expectedLockToken, actualLockToken)
+}
+
+func TestGenerateCommandId(t *testing.T) {
+
+	// Test Data
+	uid := types.UID("TestUID")
+	generation := int64(5)
+	resetOffset := &kafkav1alpha1.ResetOffset{
+		ObjectMeta: metav1.ObjectMeta{
+			UID:        uid,
+			Generation: generation,
+		},
+	}
+	podIP := "TestPodIP"
+	opCode := commands.StopConsumerGroupOpCode
+
+	// Perform The Test
+	actualCommandId, err := GenerateCommandId(resetOffset, podIP, opCode)
+
+	// Verify The Results
+	assert.Nil(t, err)
+	assert.Equal(t, int64(3448459854), actualCommandId)
+}
diff --git a/pkg/common/controlprotocol/commands/consumergroup_test.go b/pkg/common/controlprotocol/commands/consumergroup_test.go
index 463e8c58a..b87b4eeed 100644
--- a/pkg/common/controlprotocol/commands/consumergroup_test.go
+++ b/pkg/common/controlprotocol/commands/consumergroup_test.go
@@ -18,6 +18,7 @@ package commands
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -29,40 +30,89 @@ func TestNewConsumerGroupAsyncCommand(t *testing.T) {
 	topicName := "TestTopicName"
 	groupId := "TestGroupId"
 
-	// Perform The Test
-	consumerGroupAsyncCommand := NewConsumerGroupAsyncCommand(commandId, topicName, groupId)
-
-	// Verify The Results
-	assert.NotNil(t, consumerGroupAsyncCommand)
-	assert.Equal(t, ConsumerGroupAsyncCommandVersion, consumerGroupAsyncCommand.Version)
-	assert.Equal(t, commandId, consumerGroupAsyncCommand.CommandId)
-	assert.Equal(t, topicName, consumerGroupAsyncCommand.TopicName)
-	assert.Equal(t, groupId, consumerGroupAsyncCommand.GroupId)
+	// Define The Test Cases
+	tests := []struct {
+		name string
+		lock *CommandLock
+	}{
+		{
+			name: "With Lock",
+			lock: NewCommandLock("TestLockToken", 1*time.Minute, true, true),
+		},
+		{
+			name: "Without Lock",
+			lock: nil,
+		},
+	}
+
+	// Execute The Test Cases
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			// Perform The Test
+			consumerGroupAsyncCommand := NewConsumerGroupAsyncCommand(commandId, topicName, groupId, test.lock)
+
+			// Verify The Results
+			assert.NotNil(t, consumerGroupAsyncCommand)
+			assert.Equal(t, ConsumerGroupAsyncCommandVersion, consumerGroupAsyncCommand.Version)
+			assert.Equal(t, commandId, consumerGroupAsyncCommand.CommandId)
+			assert.Equal(t, topicName, consumerGroupAsyncCommand.TopicName)
+			assert.Equal(t, groupId, consumerGroupAsyncCommand.GroupId)
+			assert.Equal(t, test.lock, consumerGroupAsyncCommand.Lock)
+		})
+	}
 }
 
 func TestConsumerGroupAsyncCommand_MarshalUnmarshal(t *testing.T) {
 
-	// Create A ConsumerGroupAsyncCommand To Test
-	origConsumerGroupAsyncCommand := NewConsumerGroupAsyncCommand(int64(1234), "TestTopicName", "TestGroupId")
-
-	// Perform The Test (Marshal & Unmarshal Round Trip)
-	binaryData, err := origConsumerGroupAsyncCommand.MarshalBinary()
-	assert.Nil(t, err)
-	newConsumerGroupAsyncCommand := &ConsumerGroupAsyncCommand{}
-	err = newConsumerGroupAsyncCommand.UnmarshalBinary(binaryData)
-	assert.Nil(t, err)
-
-	// Verify The Results
-	assert.Equal(t, ConsumerGroupAsyncCommandVersion, newConsumerGroupAsyncCommand.Version)
-	assert.Equal(t, origConsumerGroupAsyncCommand.CommandId, newConsumerGroupAsyncCommand.CommandId)
-	assert.Equal(t, origConsumerGroupAsyncCommand.TopicName, newConsumerGroupAsyncCommand.TopicName)
-	assert.Equal(t, origConsumerGroupAsyncCommand.GroupId, newConsumerGroupAsyncCommand.GroupId)
+	// Test Data
+	commandId := int64(1234)
+	topicName := "TestTopicName"
+	groupId := "TestGroupId"
+
+	// Define The Test Cases
+	tests := []struct {
+		name string
+		lock *CommandLock
+	}{
+		{
+			name: "With Lock",
+			lock: NewCommandLock("TestLockToken", 1*time.Minute, true, true),
+		},
+		{
+			name: "Without Lock",
+			lock: nil,
+		},
+	}
+
+	// Execute The Test Cases
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			// Create A ConsumerGroupAsyncCommand To Test
+			origConsumerGroupAsyncCommand := NewConsumerGroupAsyncCommand(commandId, topicName, groupId, test.lock)
+
+			// Perform The Test (Marshal & Unmarshal Round Trip)
+			binaryData, err := origConsumerGroupAsyncCommand.MarshalBinary()
+			assert.Nil(t, err)
+			newConsumerGroupAsyncCommand := &ConsumerGroupAsyncCommand{}
+			err = newConsumerGroupAsyncCommand.UnmarshalBinary(binaryData)
+			assert.Nil(t, err)
+
+			// Verify The Results
+			assert.Equal(t, ConsumerGroupAsyncCommandVersion, newConsumerGroupAsyncCommand.Version)
+			assert.Equal(t, origConsumerGroupAsyncCommand.CommandId, newConsumerGroupAsyncCommand.CommandId)
+			assert.Equal(t, origConsumerGroupAsyncCommand.TopicName, newConsumerGroupAsyncCommand.TopicName)
+			assert.Equal(t, origConsumerGroupAsyncCommand.GroupId, newConsumerGroupAsyncCommand.GroupId)
+			assert.Equal(t, origConsumerGroupAsyncCommand.Lock, newConsumerGroupAsyncCommand.Lock)
+		})
+	}
 }
 
 func TestConsumerGroupAsyncCommand_SerializedId(t *testing.T) {
 
 	// Create A ConsumerGroupAsyncCommand To Test
-	consumerGroupAsyncCommand := NewConsumerGroupAsyncCommand(int64(1234), "TestTopicName", "TestGroupId1")
+	consumerGroupAsyncCommand := NewConsumerGroupAsyncCommand(int64(1234), "TestTopicName", "TestGroupId", nil)
 
 	// Perform The Test
 	serializedId := consumerGroupAsyncCommand.SerializedId()
diff --git a/pkg/common/controlprotocol/commands/lock_test.go b/pkg/common/controlprotocol/commands/lock_test.go
new file mode 100644
index 000000000..194ae0e7c
--- /dev/null
+++ b/pkg/common/controlprotocol/commands/lock_test.go
@@ -0,0 +1,40 @@
+/*
+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 commands
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewCommandLock(t *testing.T) {
+
+	// Test Data
+	lockToken := "TestLockToken"
+	lockTimeout := 1 * time.Minute
+
+	// Perform The Test
+	commandLock := NewCommandLock(lockToken, lockTimeout, true, true)
+
+	// Verify The Results
+	assert.Equal(t, lockToken, commandLock.Token)
+	assert.Equal(t, lockTimeout, commandLock.Timeout)
+	assert.True(t, commandLock.LockBefore)
+	assert.True(t, commandLock.UnlockAfter)
+}
diff --git a/pkg/common/controlprotocol/controlprotocol_test.go b/pkg/common/controlprotocol/controlprotocol_test.go
index a2da97e59..5e39a998c 100644
--- a/pkg/common/controlprotocol/controlprotocol_test.go
+++ b/pkg/common/controlprotocol/controlprotocol_test.go
@@ -50,7 +50,7 @@ func TestNewServerHandler(t *testing.T) {
 		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
-			mockService := &ctrltesting.MockControlProtocolService{}
+			mockService := &ctrltesting.MockService{}
 			mockService.On("MessageHandler", mock.Anything).Return()
 			startServerWrapper = func(_ context.Context, _ ...network.ControlServerOption) (*network.ControlServer, error) {
 				return &network.ControlServer{Service: mockService}, testCase.serverErr
@@ -70,7 +70,7 @@ func TestHandlers(t *testing.T) {
 	saveStartServer := startServerWrapper
 	defer func() { startServerWrapper = saveStartServer }()
 
-	mockService := &ctrltesting.MockControlProtocolService{}
+	mockService := &ctrltesting.MockService{}
 	mockService.On("MessageHandler", mock.Anything).Return()
 
 	startServerWrapper = func(_ context.Context, _ ...network.ControlServerOption) (*network.ControlServer, error) {
diff --git a/pkg/common/controlprotocol/testing/mocks.go b/pkg/common/controlprotocol/testing/mocks.go
index d9c2ae569..7c229bf83 100644
--- a/pkg/common/controlprotocol/testing/mocks.go
+++ b/pkg/common/controlprotocol/testing/mocks.go
@@ -22,11 +22,111 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/mock"
+	"k8s.io/apimachinery/pkg/types"
 	ctrl "knative.dev/control-protocol/pkg"
-	"knative.dev/control-protocol/pkg/message"
+	ctrlmessage "knative.dev/control-protocol/pkg/message"
+	ctrlreconciler "knative.dev/control-protocol/pkg/reconciler"
 	ctrlservice "knative.dev/control-protocol/pkg/service"
 )
 
+//
+// Mock Control-Protocol ConnectionPool
+//
+
+var _ ctrlreconciler.ControlPlaneConnectionPool = (*MockConnectionPool)(nil)
+
+type MockConnectionPool struct {
+	mock.Mock
+}
+
+func (c *MockConnectionPool) GetConnectedHosts(key string) []string {
+	args := c.Called(key)
+	return args.Get(0).([]string)
+}
+
+func (c *MockConnectionPool) GetServices(key string) map[string]ctrl.Service {
+	args := c.Called(key)
+	return args.Get(0).(map[string]ctrl.Service)
+}
+
+func (c *MockConnectionPool) ResolveControlInterface(key string, host string) (string, ctrl.Service) {
+	args := c.Called(key, host)
+	return args.String(0), args.Get(1).(ctrl.Service)
+}
+
+func (c *MockConnectionPool) RemoveConnection(ctx context.Context, key string, host string) {
+	c.Called(ctx, key, host)
+}
+
+func (c *MockConnectionPool) RemoveAllConnections(ctx context.Context, key string) {
+	c.Called(ctx, key)
+}
+
+func (c *MockConnectionPool) Close(ctx context.Context) {
+	c.Called(ctx)
+}
+
+func (c *MockConnectionPool) ReconcileConnections(ctx context.Context, key string, wantConnections []string, newServiceCb func(string, ctrl.Service), oldServiceCb func(string)) (map[string]ctrl.Service, error) {
+	args := c.Called(ctx, key, wantConnections, newServiceCb, oldServiceCb)
+	return args.Get(0).(map[string]ctrl.Service), args.Error(1)
+}
+
+func (c *MockConnectionPool) DialControlService(ctx context.Context, key string, host string) (string, ctrl.Service, error) {
+	args := c.Called(ctx, key, host)
+	return args.String(0), args.Get(1).(ctrl.Service), args.Error(2)
+}
+
+//
+// Mock Control-Protocol AsyncCommandNotificationStore
+//
+
+var _ ctrlreconciler.AsyncCommandNotificationStore = (*MockAsyncCommandNotificationStore)(nil)
+
+type MockAsyncCommandNotificationStore struct {
+	mock.Mock
+}
+
+func (a *MockAsyncCommandNotificationStore) GetCommandResult(srcName types.NamespacedName, pod string, command ctrlmessage.AsyncCommand) *ctrlmessage.AsyncCommandResult {
+	args := a.Called(srcName, pod, command)
+	return args.Get(0).(*ctrlmessage.AsyncCommandResult)
+}
+
+func (a *MockAsyncCommandNotificationStore) CleanPodsNotifications(srcName types.NamespacedName) {
+	a.Called(srcName)
+}
+
+func (a *MockAsyncCommandNotificationStore) CleanPodNotification(srcName types.NamespacedName, pod string) {
+	a.Called(srcName, pod)
+}
+
+func (a *MockAsyncCommandNotificationStore) MessageHandler(srcName types.NamespacedName, pod string) ctrl.MessageHandler {
+	args := a.Called(srcName, pod)
+	return args.Get(0).(ctrl.MessageHandler)
+}
+
+//
+// Mock Control-Protocol Service
+//
+
+var _ ctrl.Service = (*MockService)(nil)
+
+type MockService struct {
+	mock.Mock
+}
+
+func (s *MockService) SendAndWaitForAck(opcode ctrl.OpCode, payload encoding.BinaryMarshaler) error {
+	args := s.Called(opcode, payload)
+	return args.Error(0)
+}
+
+func (s *MockService) MessageHandler(handler ctrl.MessageHandler) {
+	s.Called(handler)
+}
+
+func (s *MockService) ErrorHandler(handler ctrl.ErrorHandler) {
+	s.Called(handler)
+}
+
 //
 // Mock Control-Protocol ServerHandler
 //
@@ -36,10 +136,10 @@ type MockServerHandler struct {
 	mock.Mock
 	// Export the Router field so that tests can verify handlers in it directly
 	Router  ctrlservice.MessageRouter
-	Service *MockControlProtocolService
+	Service *MockService
 }
 
-func (s *MockServerHandler) AddAsyncHandler(opcode ctrl.OpCode, resultOpcode ctrl.OpCode, payloadType message.AsyncCommand,
+func (s *MockServerHandler) AddAsyncHandler(opcode ctrl.OpCode, resultOpcode ctrl.OpCode, payloadType ctrlmessage.AsyncCommand,
 	handler func(ctx context.Context, commandMessage ctrlservice.AsyncCommandMessage)) {
 	_ = s.Called(opcode, resultOpcode, payloadType, handler)
 	s.Router[opcode] = ctrlservice.NewAsyncCommandHandler(s.Service, payloadType, resultOpcode, handler)
@@ -60,30 +160,6 @@ func (s *MockServerHandler) RemoveHandler(opcode ctrl.OpCode) {
 func GetMockServerHandler() *MockServerHandler {
 	return &MockServerHandler{
 		Router:  make(ctrlservice.MessageRouter),
-		Service: &MockControlProtocolService{},
+		Service: &MockService{},
 	}
 }
-
-//
-// Mock Control-Protocol Service
-//
-
-// MockControlProtocolService is a stub-only mock of the Control Protocol Service
-type MockControlProtocolService struct {
-	mock.Mock
-}
-
-var _ ctrl.Service = (*MockControlProtocolService)(nil)
-
-func (m *MockControlProtocolService) SendAndWaitForAck(opcode ctrl.OpCode, payload encoding.BinaryMarshaler) error {
-	args := m.Called(opcode, payload)
-	return args.Error(0)
-}
-
-func (m *MockControlProtocolService) MessageHandler(handler ctrl.MessageHandler) {
-	_ = m.Called(handler)
-}
-
-func (m *MockControlProtocolService) ErrorHandler(handler ctrl.ErrorHandler) {
-	_ = m.Called(handler)
-}