Repository: knative-sandbox/eventing-kafka
Pull request #177 (2020-11-09 21:23:29)

KafkaSource: set default zap logger level to "info"
 
<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

-
-
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
- 🧽 Default KafkaSource log level is now "info"
```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/177?src=pr&el=h1) Report
> Merging [#177](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/177?src=pr&el=desc) (fbf1a88) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/37ad805e3e591b0b51a9c20cdc0a42695a8c2102?el=desc) (37ad805) will **increase** coverage by `0.16%`.
> The diff coverage is `n/a`.

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

```diff
@@            Coverage Diff             @@
##           master     #177      +/-   ##
==========================================
+ Coverage   75.45%   75.62%   +0.16%     
==========================================
  Files         112      112              
  Lines        4140     4140              
==========================================
+ Hits         3124     3131       +7     
+ Misses        823      816       -7     
  Partials      193      193              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/177?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/177/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/177?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/177?src=pr&el=footer). Last update [37ad805...697b1a6](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/177?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/config/source/configmaps/observability.yaml b/config/source/configmaps/observability.yaml
index 62023dfb2..69cd4a5a7 100644
--- a/config/source/configmaps/observability.yaml
+++ b/config/source/configmaps/observability.yaml
@@ -22,8 +22,8 @@ data:
   # Common configuration for all Knative codebase
   zap-logger-config: |
     {
-      "level": "debug",
-      "development": true,
+      "level": "info",
+      "development": false,
       "outputPaths": ["stdout"],
       "errorOutputPaths": ["stderr"],
       "encoding": "json",
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()
+	}
 }
diff --git a/pkg/common/consumer/consumer_handler.go b/pkg/common/consumer/consumer_handler.go
index 5e1eea1a1..5cd03eb87 100644
--- a/pkg/common/consumer/consumer_handler.go
+++ b/pkg/common/consumer/consumer_handler.go
@@ -26,7 +26,7 @@ import (
 )
 
 type KafkaConsumerHandler interface {
-	// When this function returns true, the consumer group offset is committed.
+	// When this function returns true, the consumer group offset is marked as consumed.
 	// The returned error is enqueued in errors channel.
 	Handle(context context.Context, message *sarama.ConsumerMessage) (bool, error)
 }
@@ -77,7 +77,9 @@ func (consumer *SaramaConsumerHandler) ConsumeClaim(session sarama.ConsumerGroup
 			consumer.logger.Debugw("Message claimed", zap.String("topic", message.Topic), zap.Binary("value", message.Value))
 		}
 
-		mustMark, err := consumer.handler.Handle(session.Context(), message)
+		// Don't use the session context since it is closed before messages are drained.
+		// Handle must finish before the session timeout.
+		mustMark, err := consumer.handler.Handle(context.Background(), message)
 
 		if err != nil {
 			consumer.logger.Infow("Failure while handling a message", zap.String("topic", message.Topic), zap.Int32("partition", message.Partition), zap.Int64("offset", message.Offset), zap.Error(err))


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])
 }