Repository: knative-sandbox/eventing-kafka
Pull request #175 (2020-11-09 18:50:56)

Add defensive coding around reconciling deleted resources.
We were concerned that the distributed KafkaChannel could potentially hit an extremely rare (if even possible?) race condition whereby it is doing a normal Reconciliation on a resource that exists in K8S but was "being" deleted.  Therefore we'd like to add a small check for the resources Deletion Timestamp and force a re-reconciliation just in case.

> Note - The Dispatcher resources (Service / Deployment) are "cross-namespace" and thus have their own Finalizer in place and will not be modified to force re-reconciliation if the user removes the Finalizer.  Some additional state logging was added to be explicit about the handling of the resource during reconciliation.


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/175#" 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/master/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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175?src=pr&el=h1) Report
> Merging [#175](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175?src=pr&el=desc) (6b06f19) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/37ad805e3e591b0b51a9c20cdc0a42695a8c2102?el=desc) (37ad805) will **increase** coverage by `0.45%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #175      +/-   ##
==========================================
+ Coverage   75.45%   75.91%   +0.45%     
==========================================
  Files         112      112              
  Lines        4140     4164      +24     
==========================================
+ Hits         3124     3161      +37     
+ Misses        823      814       -9     
+ Partials      193      189       -4     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...nel/distributed/controller/kafkachannel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYWNoYW5uZWwvY2hhbm5lbC5nbw==) | `93.84% <100.00%> (+0.40%)` | :arrow_up: |
| [.../distributed/controller/kafkachannel/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYWNoYW5uZWwvZGlzcGF0Y2hlci5nbw==) | `90.00% <100.00%> (+2.10%)` | :arrow_up: |
| [...nel/distributed/controller/kafkasecret/receiver.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYXNlY3JldC9yZWNlaXZlci5nbw==) | `93.25% <100.00%> (+0.22%)` | :arrow_up: |
| [...g/channel/distributed/controller/util/finalizer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci91dGlsL2ZpbmFsaXplci5nbw==) | `100.00% <100.00%> (ø)` | |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `75.53% <0.00%> (+7.44%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175?src=pr&el=footer). Last update [37ad805...6b06f19](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/175?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

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/controller/kafkachannel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/175/pull-knative-sandbox-eventing-kafka-go-coverage/1325873539953201152/artifacts/line-cov.html#file0) | 91.2% | 91.9% | 0.7
[pkg/channel/distributed/controller/kafkachannel/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/175/pull-knative-sandbox-eventing-kafka-go-coverage/1325873539953201152/artifacts/line-cov.html#file1) | 84.6% | 86.4% | 1.9
[pkg/channel/distributed/controller/kafkasecret/receiver.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/175/pull-knative-sandbox-eventing-kafka-go-coverage/1325873539953201152/artifacts/line-cov.html#file3) | 86.3% | 87.3% | 1.0

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/controller/kafkachannel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/175/pull-knative-sandbox-eventing-kafka-go-coverage/1325877539847016448/artifacts/line-cov.html#file0) | 91.2% | 91.9% | 0.7
[pkg/channel/distributed/controller/kafkachannel/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/175/pull-knative-sandbox-eventing-kafka-go-coverage/1325877539847016448/artifacts/line-cov.html#file1) | 84.6% | 86.4% | 1.9
[pkg/channel/distributed/controller/kafkasecret/receiver.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/175/pull-knative-sandbox-eventing-kafka-go-coverage/1325877539847016448/artifacts/line-cov.html#file3) | 86.3% | 87.3% | 1.0

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/175#pullrequestreview-526615274" title="Approved">eric-sap</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/175#" 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/master/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/controller/kafkachannel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/175/pull-knative-sandbox-eventing-kafka-go-coverage/1325892425696153600/artifacts/line-cov.html#file0) | 91.2% | 91.9% | 0.7
[pkg/channel/distributed/controller/kafkachannel/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/175/pull-knative-sandbox-eventing-kafka-go-coverage/1325892425696153600/artifacts/line-cov.html#file1) | 84.6% | 86.4% | 1.9
[pkg/channel/distributed/controller/kafkasecret/receiver.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/175/pull-knative-sandbox-eventing-kafka-go-coverage/1325892425696153600/artifacts/line-cov.html#file3) | 86.3% | 87.3% | 1.0

phamilton: /LGTM

Patch:
diff --git a/pkg/channel/distributed/controller/kafkachannel/channel.go b/pkg/channel/distributed/controller/kafkachannel/channel.go
index 749eedace..19e76819e 100644
--- a/pkg/channel/distributed/controller/kafkachannel/channel.go
+++ b/pkg/channel/distributed/controller/kafkachannel/channel.go
@@ -66,7 +66,7 @@ func (r *Reconciler) reconcileKafkaChannelService(ctx context.Context, logger *z
 
 	// Attempt To Get The Service Associated With The Specified Channel
 	service, err := r.getKafkaChannelService(channel)
-	if err != nil {
+	if service == nil || err != nil {
 
 		// If The Service Was Not Found - Then Create A New One For The Channel
 		if errors.IsNotFound(err) {
@@ -87,8 +87,15 @@ func (r *Reconciler) reconcileKafkaChannelService(ctx context.Context, logger *z
 			return err
 		}
 	} else {
-		logger.Info("Successfully Verified KafkaChannel Service")
-		// Continue To Update Channel Status
+
+		// Verify KafkaChannel Service Is Not Terminating
+		if service.DeletionTimestamp.IsZero() {
+			logger.Info("Successfully Verified KafkaChannel Service")
+			// Continue To Update Channel Status
+		} else {
+			logger.Warn("Encountered KafkaChannel Service With DeletionTimestamp - Forcing Reconciliation", zap.String("Namespace", service.Namespace), zap.String("Name", service.Name))
+			return fmt.Errorf("encountered KafkaChannel Service with DeletionTimestamp %s/%s - potential race condition", service.Namespace, service.Name)
+		}
 	}
 
 	// Update Channel Status
diff --git a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
index 28c14709b..ab4c4e658 100644
--- a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
+++ b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
@@ -142,12 +142,18 @@ func (r *Reconciler) reconcileDispatcherService(ctx context.Context, logger *zap
 		}
 	} else {
 
-		// Check Deletion Timestamp And Log State Accordingly
-		if service.DeletionTimestamp != nil {
-			logger.Info("Blocking Pending Deletion Of Dispatcher Service")
-		} else {
+		// Log Deletion Timestamp & Finalizer State
+		if service.DeletionTimestamp.IsZero() {
 			logger.Info("Successfully Verified Dispatcher Service")
+		} else {
+			if util.HasFinalizer(r.finalizerName(), &service.ObjectMeta) {
+				logger.Info("Blocking Pending Deletion Of Dispatcher Service (Finalizer Detected)")
+			} else {
+				logger.Warn("Unable To Block Pending Deletion Of Dispatcher Service (Finalizer Missing)")
+			}
 		}
+
+		// Return Success
 		return nil
 	}
 }
@@ -288,13 +294,18 @@ func (r *Reconciler) reconcileDispatcherDeployment(ctx context.Context, logger *
 		}
 	} else {
 
-		// Check Deletion Timestamp And Log State Accordingly
-		if deployment.DeletionTimestamp != nil {
-			logger.Info("Blocking Pending Deletion Of Dispatcher Deployment")
-		} else {
+		// Log Deletion Timestamp & Finalizer State
+		if deployment.DeletionTimestamp.IsZero() {
 			logger.Info("Successfully Verified Dispatcher Deployment")
+		} else {
+			if util.HasFinalizer(r.finalizerName(), &deployment.ObjectMeta) {
+				logger.Info("Blocking Pending Deletion Of Dispatcher Deployment (Finalizer Detected)")
+			} else {
+				logger.Warn("Unable To Block Pending Deletion Of Dispatcher Deployment (Finalizer Missing)")
+			}
 		}
 
+		// Propagate Status & Return Success
 		channel.Status.PropagateDispatcherStatus(&deployment.Status)
 		return nil
 	}
diff --git a/pkg/channel/distributed/controller/kafkasecret/receiver.go b/pkg/channel/distributed/controller/kafkasecret/receiver.go
index ed1f09e85..2dcf32250 100644
--- a/pkg/channel/distributed/controller/kafkasecret/receiver.go
+++ b/pkg/channel/distributed/controller/kafkasecret/receiver.go
@@ -90,8 +90,8 @@ func (r *Reconciler) reconcileChannel(ctx context.Context, secret *corev1.Secret
 func (r *Reconciler) reconcileReceiverService(ctx context.Context, logger *zap.Logger, secret *corev1.Secret) error {
 
 	// Attempt To Get The Receiver Service Associated With The Specified Secret
-	_, err := r.getReceiverService(secret)
-	if err != nil {
+	service, err := r.getReceiverService(secret)
+	if service == nil || err != nil {
 
 		// If The Service Was Not Found - Then Create A New One For The Secret
 		if errors.IsNotFound(err) {
@@ -116,9 +116,14 @@ func (r *Reconciler) reconcileReceiverService(ctx context.Context, logger *zap.L
 		}
 	} else {
 
-		// Verified The Receiver Service Exists
-		logger.Info("Successfully Verified Receiver Service")
-		return nil
+		// Verify Receiver Service Is Not Terminating
+		if service.DeletionTimestamp.IsZero() {
+			logger.Info("Successfully Verified Receiver Service")
+			return nil
+		} else {
+			logger.Warn("Encountered Receiver Service With DeletionTimestamp - Forcing Reconciliation", zap.String("Namespace", service.Namespace), zap.String("Name", service.Name))
+			return fmt.Errorf("encountered Receiver Service with DeletionTimestamp %s/%s - potential race condition", service.Namespace, service.Name)
+		}
 	}
 }
 
@@ -186,8 +191,8 @@ func (r *Reconciler) newReceiverService(secret *corev1.Secret) *corev1.Service {
 func (r *Reconciler) reconcileReceiverDeployment(ctx context.Context, logger *zap.Logger, secret *corev1.Secret) error {
 
 	// Attempt To Get The Receiver Deployment Associated With The Specified Secret
-	_, err := r.getReceiverDeployment(secret)
-	if err != nil {
+	deployment, err := r.getReceiverDeployment(secret)
+	if deployment == nil || err != nil {
 
 		// If The Receiver Deployment Was Not Found - Then Create A New Deployment For The Secret
 		if errors.IsNotFound(err) {
@@ -217,9 +222,14 @@ func (r *Reconciler) reconcileReceiverDeployment(ctx context.Context, logger *za
 		}
 	} else {
 
-		// Verified The Receiver Deployment Exists
-		logger.Info("Successfully Verified Receiver Deployment")
-		return nil
+		// Verify Receiver Deployment Is Not Terminating
+		if deployment.DeletionTimestamp.IsZero() {
+			logger.Info("Successfully Verified Receiver Deployment")
+			return nil
+		} else {
+			logger.Warn("Encountered Receiver Deployment With DeletionTimestamp - Forcing Reconciliation", zap.String("Namespace", deployment.Namespace), zap.String("Name", deployment.Name))
+			return fmt.Errorf("encountered Receiver Deployment with DeletionTimestamp %s/%s - potential race condition", deployment.Namespace, deployment.Name)
+		}
 	}
 }
 
diff --git a/pkg/channel/distributed/controller/util/finalizer.go b/pkg/channel/distributed/controller/util/finalizer.go
index 5a896b933..2dc4f24be 100644
--- a/pkg/channel/distributed/controller/util/finalizer.go
+++ b/pkg/channel/distributed/controller/util/finalizer.go
@@ -45,9 +45,21 @@ func KubernetesResourceFinalizerName(finalizerSuffix string) string {
 	return constants.EventingKafkaFinalizerPrefix + finalizerSuffix
 }
 
+// Determine Whether ObjectMeta Contains Specified Finalizer
+func HasFinalizer(finalizer string, objectMeta *metav1.ObjectMeta) bool {
+	if objectMeta != nil {
+		finalizers := sets.NewString(objectMeta.Finalizers...)
+		return finalizers.Has(finalizer)
+	} else {
+		return false
+	}
+}
+
 // Remove The Specified Finalizer From The Object
-func RemoveFinalizer(finalizer string, obj *metav1.ObjectMeta) {
-	finalizers := sets.NewString(obj.Finalizers...)
-	finalizers.Delete(finalizer)
-	obj.Finalizers = finalizers.List()
+func RemoveFinalizer(finalizer string, objectMeta *metav1.ObjectMeta) {
+	if objectMeta != nil {
+		finalizers := sets.NewString(objectMeta.Finalizers...)
+		finalizers.Delete(finalizer)
+		objectMeta.Finalizers = finalizers.List()
+	}
 }


Test patch:
diff --git a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
index c95beed13..4d893b310 100644
--- a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
@@ -321,6 +321,32 @@ func TestReconcile(t *testing.T) {
 				controllertesting.NewKafkaChannelFailedReconciliationEvent(),
 			},
 		},
+		{
+			Name:                    "Reconcile KafkaChannel Service With Deletion Timestamp",
+			SkipNamespaceValidation: true,
+			Key:                     controllertesting.KafkaChannelKey,
+			Objects: []runtime.Object{
+				controllertesting.NewKafkaChannel(
+					controllertesting.WithFinalizer,
+					controllertesting.WithMetaData,
+					controllertesting.WithAddress,
+					controllertesting.WithInitializedConditions,
+					controllertesting.WithKafkaChannelServiceReady,
+					controllertesting.WithDispatcherDeploymentReady,
+					controllertesting.WithTopicReady,
+				),
+				controllertesting.NewKafkaChannelService(controllertesting.WithDeletionTimestampService),
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
+				controllertesting.NewKafkaChannelDispatcherService(),
+				controllertesting.NewKafkaChannelDispatcherDeployment(),
+			},
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(corev1.EventTypeWarning, event.KafkaChannelServiceReconciliationFailed.String(), "Failed To Reconcile KafkaChannel Service: encountered KafkaChannel Service with DeletionTimestamp kafkachannel-namespace/kafkachannel-name-kn-channel - potential race condition"),
+				controllertesting.NewKafkaChannelFailedReconciliationEvent(),
+			},
+		},
 
 		//
 		// KafkaChannel Dispatcher Service
@@ -382,6 +408,56 @@ func TestReconcile(t *testing.T) {
 				controllertesting.NewKafkaChannelFailedReconciliationEvent(),
 			},
 		},
+		{
+			Name:                    "Reconcile Dispatcher Service With Deletion Timestamp And Finalizer",
+			SkipNamespaceValidation: true,
+			Key:                     controllertesting.KafkaChannelKey,
+			Objects: []runtime.Object{
+				controllertesting.NewKafkaChannel(
+					controllertesting.WithFinalizer,
+					controllertesting.WithMetaData,
+					controllertesting.WithAddress,
+					controllertesting.WithInitializedConditions,
+					controllertesting.WithKafkaChannelServiceReady,
+					controllertesting.WithDispatcherDeploymentReady,
+					controllertesting.WithTopicReady,
+				),
+				controllertesting.NewKafkaChannelService(),
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
+				controllertesting.NewKafkaChannelDispatcherService(controllertesting.WithDeletionTimestampService),
+				controllertesting.NewKafkaChannelDispatcherDeployment(),
+			},
+			WantErr: false,
+			WantEvents: []string{
+				controllertesting.NewKafkaChannelSuccessfulReconciliationEvent(),
+			},
+		},
+		{
+			Name:                    "Reconcile Dispatcher Service With Deletion Timestamp And Missing Finalizer",
+			SkipNamespaceValidation: true,
+			Key:                     controllertesting.KafkaChannelKey,
+			Objects: []runtime.Object{
+				controllertesting.NewKafkaChannel(
+					controllertesting.WithFinalizer,
+					controllertesting.WithMetaData,
+					controllertesting.WithAddress,
+					controllertesting.WithInitializedConditions,
+					controllertesting.WithKafkaChannelServiceReady,
+					controllertesting.WithDispatcherDeploymentReady,
+					controllertesting.WithTopicReady,
+				),
+				controllertesting.NewKafkaChannelService(),
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
+				controllertesting.NewKafkaChannelDispatcherService(controllertesting.WithoutFinalizersService, controllertesting.WithDeletionTimestampService),
+				controllertesting.NewKafkaChannelDispatcherDeployment(),
+			},
+			WantErr: false,
+			WantEvents: []string{
+				controllertesting.NewKafkaChannelSuccessfulReconciliationEvent(),
+			},
+		},
 
 		//
 		// KafkaChannel Dispatcher Deployment
@@ -455,6 +531,56 @@ func TestReconcile(t *testing.T) {
 				controllertesting.NewKafkaChannelFailedReconciliationEvent(),
 			},
 		},
+		{
+			Name:                    "Reconcile Dispatcher Deployment With Deletion Timestamp And Finalizer",
+			SkipNamespaceValidation: true,
+			Key:                     controllertesting.KafkaChannelKey,
+			Objects: []runtime.Object{
+				controllertesting.NewKafkaChannel(
+					controllertesting.WithFinalizer,
+					controllertesting.WithMetaData,
+					controllertesting.WithAddress,
+					controllertesting.WithInitializedConditions,
+					controllertesting.WithKafkaChannelServiceReady,
+					controllertesting.WithDispatcherDeploymentReady,
+					controllertesting.WithTopicReady,
+				),
+				controllertesting.NewKafkaChannelService(),
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
+				controllertesting.NewKafkaChannelDispatcherService(),
+				controllertesting.NewKafkaChannelDispatcherDeployment(controllertesting.WithDeletionTimestampDeployment),
+			},
+			WantErr: false,
+			WantEvents: []string{
+				controllertesting.NewKafkaChannelSuccessfulReconciliationEvent(),
+			},
+		},
+		{
+			Name:                    "Reconcile Dispatcher Deployment With Deletion Timestamp And Missing Finalizer",
+			SkipNamespaceValidation: true,
+			Key:                     controllertesting.KafkaChannelKey,
+			Objects: []runtime.Object{
+				controllertesting.NewKafkaChannel(
+					controllertesting.WithFinalizer,
+					controllertesting.WithMetaData,
+					controllertesting.WithAddress,
+					controllertesting.WithInitializedConditions,
+					controllertesting.WithKafkaChannelServiceReady,
+					controllertesting.WithDispatcherDeploymentReady,
+					controllertesting.WithTopicReady,
+				),
+				controllertesting.NewKafkaChannelService(),
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
+				controllertesting.NewKafkaChannelDispatcherService(),
+				controllertesting.NewKafkaChannelDispatcherDeployment(controllertesting.WithoutFinalizersDeployment, controllertesting.WithDeletionTimestampDeployment),
+			},
+			WantErr: false,
+			WantEvents: []string{
+				controllertesting.NewKafkaChannelSuccessfulReconciliationEvent(),
+			},
+		},
 	}
 
 	// Mock The Common Kafka AdminClient Creation For Test
diff --git a/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go b/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go
index a88973dd1..cd07c8430 100644
--- a/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go
+++ b/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go
@@ -194,6 +194,20 @@ func TestReconcile(t *testing.T) {
 				controllertesting.NewKafkaSecretFailedReconciliationEvent(),
 			},
 		},
+		{
+			Name: "Reconcile Receiver Service With Deletion Timestamp",
+			Key:  controllertesting.KafkaSecretKey,
+			Objects: []runtime.Object{
+				controllertesting.NewKafkaSecret(controllertesting.WithKafkaSecretFinalizer),
+				controllertesting.NewKafkaChannelReceiverService(controllertesting.WithDeletionTimestampService),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
+			},
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(corev1.EventTypeWarning, event.ReceiverServiceReconciliationFailed.String(), "Failed To Reconcile Receiver Service: encountered Receiver Service with DeletionTimestamp knative-eventing/kafkasecret-name-b9176d5f-receiver - potential race condition"),
+				controllertesting.NewKafkaSecretFailedReconciliationEvent(),
+			},
+		},
 
 		//
 		// KafkaSecret Receiver Deployment
@@ -242,6 +256,20 @@ func TestReconcile(t *testing.T) {
 				controllertesting.NewKafkaSecretFailedReconciliationEvent(),
 			},
 		},
+		{
+			Name: "Reconcile Receiver Deployment With Deletion Timestamp",
+			Key:  controllertesting.KafkaSecretKey,
+			Objects: []runtime.Object{
+				controllertesting.NewKafkaSecret(controllertesting.WithKafkaSecretFinalizer),
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(controllertesting.WithDeletionTimestampDeployment),
+			},
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(corev1.EventTypeWarning, event.ReceiverDeploymentReconciliationFailed.String(), "Failed To Reconcile Receiver Deployment: encountered Receiver Deployment with DeletionTimestamp knative-eventing/kafkasecret-name-b9176d5f-receiver - potential race condition"),
+				controllertesting.NewKafkaSecretFailedReconciliationEvent(),
+			},
+		},
 	}
 
 	// Run The TableTest Using The KafkaChannel Reconciler Provided By The Factory
diff --git a/pkg/channel/distributed/controller/testing/data.go b/pkg/channel/distributed/controller/testing/data.go
index 31208abb8..9be943507 100644
--- a/pkg/channel/distributed/controller/testing/data.go
+++ b/pkg/channel/distributed/controller/testing/data.go
@@ -19,7 +19,6 @@ package testing
 import (
 	"fmt"
 	"strconv"
-	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -152,6 +151,7 @@ Producer:
 
 var (
 	DefaultRetentionMillisString = strconv.FormatInt(DefaultRetentionMillis, 10)
+	DeletionTimestamp            = metav1.Now()
 )
 
 //
@@ -162,6 +162,16 @@ var (
 type ServiceOption func(service *corev1.Service)
 type DeploymentOption func(service *appsv1.Deployment)
 
+// Set The Service's DeletionTimestamp To Current Time
+func WithDeletionTimestampService(service *corev1.Service) {
+	service.ObjectMeta.SetDeletionTimestamp(&DeletionTimestamp)
+}
+
+// Set The Deployment's DeletionTimestamp To Current Time
+func WithDeletionTimestampDeployment(deployment *appsv1.Deployment) {
+	deployment.ObjectMeta.SetDeletionTimestamp(&DeletionTimestamp)
+}
+
 // Clear The Specified Service's Finalizers
 func WithoutFinalizersService(service *corev1.Service) {
 	service.ObjectMeta.Finalizers = []string{}
@@ -258,8 +268,7 @@ func NewKafkaSecret(options ...KafkaSecretOption) *corev1.Secret {
 
 // Set The Kafka Secret's DeletionTimestamp To Current Time
 func WithKafkaSecretDeleted(secret *corev1.Secret) {
-	deleteTime := metav1.NewTime(time.Unix(1e9, 0))
-	secret.ObjectMeta.SetDeletionTimestamp(&deleteTime)
+	secret.ObjectMeta.SetDeletionTimestamp(&DeletionTimestamp)
 }
 
 // Set The Kafka Secret's Finalizer
@@ -346,8 +355,7 @@ func WithInitializedConditions(kafkachannel *kafkav1beta1.KafkaChannel) {
 
 // Set The KafkaChannel's DeletionTimestamp To Current Time
 func WithDeletionTimestamp(kafkachannel *kafkav1beta1.KafkaChannel) {
-	deleteTime := metav1.NewTime(time.Unix(1e9, 0))
-	kafkachannel.ObjectMeta.SetDeletionTimestamp(&deleteTime)
+	kafkachannel.ObjectMeta.SetDeletionTimestamp(&DeletionTimestamp)
 }
 
 // Set The KafkaChannel's Finalizer
@@ -442,8 +450,10 @@ func WithTopicReady(kafkachannel *kafkav1beta1.KafkaChannel) {
 }
 
 // Utility Function For Creating A Custom KafkaChannel "Channel" Service For Testing
-func NewKafkaChannelService() *corev1.Service {
-	return &corev1.Service{
+func NewKafkaChannelService(options ...ServiceOption) *corev1.Service {
+
+	// Create The KafkaChannel Service
+	service := &corev1.Service{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: corev1.SchemeGroupVersion.String(),
 			Kind:       constants.ServiceKind,
@@ -466,11 +476,21 @@ func NewKafkaChannelService() *corev1.Service {
 			ExternalName: ReceiverDeploymentName + "." + commonconstants.KnativeEventingNamespace + ".svc.cluster.local",
 		},
 	}
+
+	// Apply The Specified Service Customizations
+	for _, option := range options {
+		option(service)
+	}
+
+	// Return The Test KafkaChannel Service
+	return service
 }
 
 // Utility Function For Creating A Custom Receiver Service For Testing
-func NewKafkaChannelReceiverService() *corev1.Service {
-	return &corev1.Service{
+func NewKafkaChannelReceiverService(options ...ServiceOption) *corev1.Service {
+
+	// Create The Receiver Service
+	service := &corev1.Service{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: corev1.SchemeGroupVersion.String(),
 			Kind:       constants.ServiceKind,
@@ -504,12 +524,24 @@ func NewKafkaChannelReceiverService() *corev1.Service {
 			},
 		},
 	}
+
+	// Apply The Specified Service Customizations
+	for _, option := range options {
+		option(service)
+	}
+
+	// Return The Test Receiver Service
+	return service
 }
 
 // Utility Function For Creating A Receiver Deployment For The Test Channel
-func NewKafkaChannelReceiverDeployment() *appsv1.Deployment {
+func NewKafkaChannelReceiverDeployment(options ...DeploymentOption) *appsv1.Deployment {
+
+	// Replicas Int Reference
 	replicas := int32(ReceiverReplicas)
-	return &appsv1.Deployment{
+
+	// Create The Receiver Deployment
+	deployment := &appsv1.Deployment{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: appsv1.SchemeGroupVersion.String(),
 			Kind:       constants.DeploymentKind,
@@ -652,6 +684,14 @@ func NewKafkaChannelReceiverDeployment() *appsv1.Deployment {
 			},
 		},
 	}
+
+	// Apply The Specified Deployment Customizations
+	for _, option := range options {
+		option(deployment)
+	}
+
+	// Return The Test Receiver Deployment
+	return deployment
 }
 
 // Utility Function For Creating A Custom KafkaChannel Dispatcher Service For Testing
diff --git a/pkg/channel/distributed/controller/util/finalizer_test.go b/pkg/channel/distributed/controller/util/finalizer_test.go
index 1b962c00c..2189b2497 100644
--- a/pkg/channel/distributed/controller/util/finalizer_test.go
+++ b/pkg/channel/distributed/controller/util/finalizer_test.go
@@ -31,6 +31,25 @@ func TestKubernetesResourceFinalizerName(t *testing.T) {
 	assert.Equal(t, constants.EventingKafkaFinalizerPrefix+suffix, result)
 }
 
+// Test The HasFinalizer() Functionality
+func TestHasFinalizer(t *testing.T) {
+
+	// Test Data
+	finalizer1 := "TestFinalizer1"
+	finalizer2 := "TestFinalizer2"
+	finalizer3 := "TestFinalizer3"
+	objectMeta := &metav1.ObjectMeta{
+		Finalizers: []string{finalizer1, finalizer2},
+	}
+
+	// Perform The Test
+	assert.False(t, HasFinalizer(finalizer1, nil))
+	assert.False(t, HasFinalizer(finalizer1, &metav1.ObjectMeta{}))
+	assert.True(t, HasFinalizer(finalizer1, objectMeta))
+	assert.True(t, HasFinalizer(finalizer2, objectMeta))
+	assert.False(t, HasFinalizer(finalizer3, objectMeta))
+}
+
 // Test The RemoveFinalizer() Functionality
 func TestRemoveFinalizer(t *testing.T) {
 
@@ -38,15 +57,15 @@ func TestRemoveFinalizer(t *testing.T) {
 	finalizer1 := "TestFinalizer1"
 	finalizer2 := "TestFinalizer2"
 	finalizer3 := "TestFinalizer3"
-	objMeta := &metav1.ObjectMeta{
+	objectMeta := &metav1.ObjectMeta{
 		Finalizers: []string{finalizer1, finalizer2, finalizer3},
 	}
 
 	// Perform The Test
-	RemoveFinalizer(finalizer2, objMeta)
+	RemoveFinalizer(finalizer2, objectMeta)
 
 	// Verify The Results
-	assert.Equal(t, 2, len(objMeta.Finalizers))
-	assert.Equal(t, finalizer1, objMeta.Finalizers[0])
-	assert.Equal(t, finalizer3, objMeta.Finalizers[1])
+	assert.Equal(t, 2, len(objectMeta.Finalizers))
+	assert.Equal(t, finalizer1, objectMeta.Finalizers[0])
+	assert.Equal(t, finalizer3, objectMeta.Finalizers[1])
 }