Repository: knative-sandbox/eventing-kafka
Pull request #94 (2020-10-12 16:26:17)

refactor kafka producer from being called a channel to a receiver.
Recently, in the distributed kafka channel implementation, we renamed the Kafka Producer pod's top level directory from `channel/` to `receiver` as the term "channel" was being abused/overused.  This PR finishes that renaming effort by changing the code to read consistently, and altering the runtime artifacts naming to include `receiver` instead of `channel`.  Other than naming, and some other general code clean-up, there are no intended behavioural differences.


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/94#" 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/kafkasecret/receiver.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/94/pull-knative-sandbox-eventing-kafka-go-coverage/1315690466171162626/artifacts/line-cov.html#file4) | Do not exist | 86.3% | 

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/kafkasecret/receiver.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/94/pull-knative-sandbox-eventing-kafka-go-coverage/1315699525762945026/artifacts/line-cov.html#file4) | Do not exist | 86.3% | 

travis-minke-sap: /hold

found additional touch-points
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/kafkasecret/receiver.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/94/pull-knative-sandbox-eventing-kafka-go-coverage/1315760678765596678/artifacts/line-cov.html#file12) | Do not exist | 86.3% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-tests | 893d3c5ed8357e0803b0b7895306075329e1a6f6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/94/pull-knative-sandbox-eventing-kafka-integration-tests/1315760678765596680) | `/test pull-knative-sandbox-eventing-kafka-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/94). [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-integration-tests
travis-minke-sap: /unhold

Patch:
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index 9670ff346..64c33f67f 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -14,6 +14,7 @@ import (
 	commonk8s "knative.dev/eventing-kafka/pkg/channel/distributed/common/k8s"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/metrics"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/controller"
 	dispatch "knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/dispatcher"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/env"
@@ -26,11 +27,6 @@ import (
 	"knative.dev/pkg/signals"
 )
 
-// Constants
-const (
-	Component = "KafkaDispatcher"
-)
-
 // Variables
 var (
 	logger     *zap.Logger
@@ -46,7 +42,7 @@ func main() {
 	flag.Parse()
 
 	// Initialize A Knative Injection Lite Context (K8S Client & Logger)
-	ctx := commonk8s.LoggingContext(signals.NewContext(), Component, *serverURL, *kubeconfig)
+	ctx := commonk8s.LoggingContext(signals.NewContext(), constants.Component, *serverURL, *kubeconfig)
 
 	// Get The Logger From The Context & Defer Flushing Any Buffered Log Entries On Exit
 	logger = logging.FromContext(ctx).Desugar()
@@ -68,7 +64,7 @@ func main() {
 	}
 
 	// Update The Sarama Config - Username/Password Overrides (EnvVars From Secret Take Precedence Over ConfigMap)
-	sarama.UpdateSaramaConfig(saramaConfig, Component, environment.KafkaUsername, environment.KafkaPassword)
+	sarama.UpdateSaramaConfig(saramaConfig, constants.Component, environment.KafkaUsername, environment.KafkaPassword)
 
 	// Initialize Tracing (Watches config-tracing ConfigMap, Assumes Context Came From LoggingContext With Embedded K8S Client Key)
 	err = commonconfig.InitializeTracing(logger.Sugar(), ctx, environment.ServiceName)
@@ -91,7 +87,7 @@ func main() {
 	// Create The Dispatcher With Specified Configuration
 	dispatcherConfig := dispatch.DispatcherConfig{
 		Logger:        logger,
-		ClientId:      Component,
+		ClientId:      constants.Component,
 		Brokers:       strings.Split(environment.KafkaBrokers, ","),
 		Topic:         environment.KafkaTopic,
 		Username:      environment.KafkaUsername,
diff --git a/cmd/channel/distributed/receiver/main.go b/cmd/channel/distributed/receiver/main.go
index 51d48a2e3..b41de175e 100644
--- a/cmd/channel/distributed/receiver/main.go
+++ b/cmd/channel/distributed/receiver/main.go
@@ -25,11 +25,6 @@ import (
 	eventingmetrics "knative.dev/pkg/metrics"
 )
 
-// Constants
-const (
-	Component = "KafkaChannel"
-)
-
 // Variables
 var (
 	logger        *zap.Logger
@@ -67,7 +62,7 @@ func main() {
 	}
 
 	// Update The Sarama Config - Username/Password Overrides (EnvVars From Secret Take Precedence Over ConfigMap)
-	sarama.UpdateSaramaConfig(saramaConfig, Component, environment.KafkaUsername, environment.KafkaPassword)
+	sarama.UpdateSaramaConfig(saramaConfig, constants.Component, environment.KafkaUsername, environment.KafkaPassword)
 
 	// Initialize Tracing (Watches config-tracing ConfigMap, Assumes Context Came From LoggingContext With Embedded K8S Client Key)
 	err = commonconfig.InitializeTracing(logger.Sugar(), ctx, environment.ServiceName)
diff --git a/config/channel/distributed/200-eventing-kafka-configmap.yaml b/config/channel/distributed/200-eventing-kafka-configmap.yaml
index 6dc1b350f..59339ebd1 100644
--- a/config/channel/distributed/200-eventing-kafka-configmap.yaml
+++ b/config/channel/distributed/200-eventing-kafka-configmap.yaml
@@ -24,7 +24,7 @@ data:
       Idempotent: true  # Must be false for Azure EventHubs
       RequiredAcks: -1  # -1 = WaitForAll, Most stringent option for "at-least-once" delivery.
   eventing-kafka: |
-    channel:
+    receiver:
       cpuLimit: 200m
       cpuRequest: 100m
       memoryLimit: 100Mi
diff --git a/config/channel/distributed/400-deployment.yaml b/config/channel/distributed/400-deployment.yaml
index 040275d03..392c9c57f 100644
--- a/config/channel/distributed/400-deployment.yaml
+++ b/config/channel/distributed/400-deployment.yaml
@@ -47,7 +47,7 @@ spec:
           value: "8081"
         - name: METRICS_DOMAIN
           value: "eventing-kafka"
-        - name: CHANNEL_IMAGE
+        - name: RECEIVER_IMAGE
           value: "ko://knative.dev/eventing-kafka/cmd/channel/distributed/receiver"
         - name: DISPATCHER_IMAGE
           value: "ko://knative.dev/eventing-kafka/cmd/channel/distributed/dispatcher"
diff --git a/config/channel/distributed/README.md b/config/channel/distributed/README.md
index e9911de09..287c40fdd 100644
--- a/config/channel/distributed/README.md
+++ b/config/channel/distributed/README.md
@@ -127,7 +127,7 @@ cluster.
 
 - **eventing-kafka:** This section provides customization of runtime behavior of the eventing-kafka implementation as follows...
 
-  - **channel:** Controls the Deployment runtime characteristics of the Channel (one Deployment per Kafka Secret).
+  - **receiver:** Controls the Deployment runtime characteristics of the Receiver (one Deployment per Kafka Secret).
   - **dispatcher:** Controls the Deployment runtime characterstics of the Dispatcher (one Deployment per KafkaChannel CR).
   - **kafka.defaultReplicationFactor:** Cannot exceed the number of Kafka Brokers configured in your system.
   - **kafka.adminType:** As described above this value must be set to one of `kafka`, `azure`, or `custom`.  The default is `kakfa` and will be used by most users.
diff --git a/config/source/configmaps/observability.yaml b/config/source/configmaps/observability.yaml
index a56fd5d0b..1ec1416fc 100644
--- a/config/source/configmaps/observability.yaml
+++ b/config/source/configmaps/observability.yaml
@@ -189,7 +189,7 @@ apiVersion: v1
 kind: ConfigMap
 metadata:
   name: config-tracing
-  namespace: knative-eventing
+  namespace: knative-sources
   labels:
     eventing.knative.dev/release: devel
     knative.dev/config-propagation: original
diff --git a/pkg/channel/distributed/README.md b/pkg/channel/distributed/README.md
index 1ffaf0cc5..12d027639 100644
--- a/pkg/channel/distributed/README.md
+++ b/pkg/channel/distributed/README.md
@@ -1,6 +1,6 @@
-# Contributed Kafka Channel implementation
+# Eventing-Kafka Distributed Channel
 
-This is a Kafka Channel implementation, contributed by
+This is a Kafka Channel implementation, originally contributed by
 [SAP's Kyma project](https://github.com/kyma-project).
 
 > See <https://github.com/knative/eventing-contrib/issues/1070> for discussion of the donation process.
diff --git a/pkg/channel/distributed/common/config/configwatcher.go b/pkg/channel/distributed/common/config/configwatcher.go
index 7c6f69b2b..1560164d8 100644
--- a/pkg/channel/distributed/common/config/configwatcher.go
+++ b/pkg/channel/distributed/common/config/configwatcher.go
@@ -24,8 +24,8 @@ type EKKubernetesConfig struct {
 	Replicas      int               `json:"replicas,omitempty"`
 }
 
-// The Channel config has nothing in it except the base Kubernetes fields (Cpu, Memory, Replicas)
-type EKChannelConfig struct {
+// The Receiver config has nothing in it except the base Kubernetes fields (Cpu, Memory, Replicas)
+type EKReceiverConfig struct {
 	EKKubernetesConfig
 }
 
@@ -47,9 +47,9 @@ type EKKafkaConfig struct {
 	AdminType string             `json:"adminType,omitempty"`
 }
 
-// EventingKafkaConfig is the main struct that holds the Channel, Dispatcher, and Kafka sub-items
+// EventingKafkaConfig is the main struct that holds the Receiver, Dispatcher, and Kafka sub-items
 type EventingKafkaConfig struct {
-	Channel    EKChannelConfig    `json:"channel,omitempty"`
+	Receiver   EKReceiverConfig   `json:"receiver,omitempty"`
 	Dispatcher EKDispatcherConfig `json:"dispatcher,omitempty"`
 	Kafka      EKKafkaConfig      `json:"kafka,omitempty"`
 }
diff --git a/pkg/channel/distributed/common/k8s/context.go b/pkg/channel/distributed/common/k8s/context.go
index 6657e418e..1ac593b12 100644
--- a/pkg/channel/distributed/common/k8s/context.go
+++ b/pkg/channel/distributed/common/k8s/context.go
@@ -33,7 +33,7 @@ var K8sClientWrapper = func(serverUrl string, kubeconfigPath string) kubernetes.
 // Initialize The Specified Context With A K8S Client & Logger (ConfigMap Watcher)
 //
 // Note - This logic represents a stepping stone on our path towards alignment with the Knative eventing-contrib implementations.
-//        The Channel / Dispatcher are not "injected controllers" in the knative-eventing injection framework, but still want to
+//        The Receiver / Dispatcher are not "injected controllers" in the knative-eventing injection framework, but still want to
 //        leverage that implementation as much as possible to ease future refactoring.  This will allow us to use the default
 //        knative-eventing logging configuration and dynamic updating.  To that end, we are setting up a basic context ourselves
 //        that mirrors what the injection framework would have created.
diff --git a/pkg/channel/distributed/controller/config/config.go b/pkg/channel/distributed/controller/config/config.go
index a663edf41..4a1337e7f 100644
--- a/pkg/channel/distributed/controller/config/config.go
+++ b/pkg/channel/distributed/controller/config/config.go
@@ -47,16 +47,16 @@ func VerifyConfiguration(configuration *config.EventingKafkaConfig) error {
 		return ControllerConfigurationError("Dispatcher.MemoryRequest must be nonzero")
 	case configuration.Dispatcher.Replicas < 1:
 		return ControllerConfigurationError("Dispatcher.Replicas must be > 0")
-	case configuration.Channel.CpuLimit == resource.Quantity{}:
-		return ControllerConfigurationError("Channel.CpuLimit must be nonzero")
-	case configuration.Channel.CpuRequest == resource.Quantity{}:
-		return ControllerConfigurationError("Channel.CpuRequest must be nonzero")
-	case configuration.Channel.MemoryLimit == resource.Quantity{}:
-		return ControllerConfigurationError("Channel.MemoryLimit must be nonzero")
-	case configuration.Channel.MemoryRequest == resource.Quantity{}:
-		return ControllerConfigurationError("Channel.MemoryRequest must be nonzero")
-	case configuration.Channel.Replicas < 1:
-		return ControllerConfigurationError("Channel.Replicas must be > 0")
+	case configuration.Receiver.CpuLimit == resource.Quantity{}:
+		return ControllerConfigurationError("Receiver.CpuLimit must be nonzero")
+	case configuration.Receiver.CpuRequest == resource.Quantity{}:
+		return ControllerConfigurationError("Receiver.CpuRequest must be nonzero")
+	case configuration.Receiver.MemoryLimit == resource.Quantity{}:
+		return ControllerConfigurationError("Receiver.MemoryLimit must be nonzero")
+	case configuration.Receiver.MemoryRequest == resource.Quantity{}:
+		return ControllerConfigurationError("Receiver.MemoryRequest must be nonzero")
+	case configuration.Receiver.Replicas < 1:
+		return ControllerConfigurationError("Receiver.Replicas must be > 0")
 	}
 	return nil // no problems found
 }
diff --git a/pkg/channel/distributed/controller/constants/constants.go b/pkg/channel/distributed/controller/constants/constants.go
index 6bc83b08a..90cc33cb9 100644
--- a/pkg/channel/distributed/controller/constants/constants.go
+++ b/pkg/channel/distributed/controller/constants/constants.go
@@ -8,15 +8,11 @@ const (
 	KafkaAdminTypeValueCustom = "custom"
 
 	// The Controller's Component Name (Needs To Be DNS Safe!)
-	ControllerComponentName = "eventingkafka-controller"
+	ControllerComponentName = "eventing-kafka-channel-controller"
 
 	// Knative Duck Versions
 	SubscribableDuckVersionAnnotationV1 = "v1"
 
-	// Knative Controller Naming
-	KafkaChannelControllerAgentName = "kafka-channel-controller"
-	KafkaSecretControllerAgentName  = "kafka-secret-controller"
-
 	// CRD Kinds
 	SecretKind              = "Secret"
 	ServiceKind             = "Service"
@@ -53,7 +49,7 @@ const (
 	AppLabel                    = "app"
 	KafkaChannelNameLabel       = "kafkachannel-name"
 	KafkaChannelNamespaceLabel  = "kafkachannel-namespace"
-	KafkaChannelChannelLabel    = "kafkachannel-channel"    // Channel Label - Used To Mark Deployment As Related To Channel
+	KafkaChannelReceiverLabel   = "kafkachannel-receiver"   // Receiver Label - Used To Mark Deployment As Receiver
 	KafkaChannelDispatcherLabel = "kafkachannel-dispatcher" // Dispatcher Label - Used To Mark Deployment As Dispatcher
 	KafkaSecretLabel            = "kafkasecret"             // Secret Label - Indicates The Kafka Secret Of The KafkaChannel
 	KafkaTopicLabel             = "kafkaTopic"              // Topic Label - Indicates The Kafka Topic Of The KnativeChannel
diff --git a/pkg/channel/distributed/controller/env/environment.go b/pkg/channel/distributed/controller/env/environment.go
index 02a65697b..313159450 100644
--- a/pkg/channel/distributed/controller/env/environment.go
+++ b/pkg/channel/distributed/controller/env/environment.go
@@ -8,11 +8,11 @@ import (
 // Package Constants
 const (
 
-	// Dispatcher Resources
+	// Dispatcher Configuration
 	DispatcherImageEnvVarKey = "DISPATCHER_IMAGE"
 
-	// Channel Resources
-	ChannelImageEnvVarKey = "CHANNEL_IMAGE"
+	// Receiver Configuration
+	ReceiverImageEnvVarKey = "RECEIVER_IMAGE"
 )
 
 // Environment Structure
@@ -23,11 +23,11 @@ type Environment struct {
 	MetricsPort    int    // Required
 	MetricsDomain  string // Required
 
-	// Resource configuration
+	// Dispatcher Configuration
 	DispatcherImage string // Required
 
-	// Resource Limits for each Channel Deployment
-	ChannelImage string // Required
+	// Receiver Configuration
+	ReceiverImage string // Required
 }
 
 // Get The Environment
@@ -68,11 +68,11 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	}
 
 	//
-	// Channel Configuration
+	// Receiver Configuration
 	//
 
-	// Get The Required ChannelImage Config Value
-	environment.ChannelImage, err = env.GetRequiredConfigValue(logger, ChannelImageEnvVarKey)
+	// Get The Required ReceiverImage Config Value
+	environment.ReceiverImage, err = env.GetRequiredConfigValue(logger, ReceiverImageEnvVarKey)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/channel/distributed/controller/event/event.go b/pkg/channel/distributed/controller/event/event.go
index b525de1de..05b47af58 100644
--- a/pkg/channel/distributed/controller/event/event.go
+++ b/pkg/channel/distributed/controller/event/event.go
@@ -18,12 +18,12 @@ const (
 
 	// Channel Updates (Finalizers, Status)
 	ChannelUpdateFailed
-
-	// Channel (Kafka Producer) Reconciliation
-	ChannelServiceReconciliationFailed
-	ChannelDeploymentReconciliationFailed
 	ChannelStatusReconciliationFailed
 
+	// Receiver (Kafka Producer) Reconciliation
+	ReceiverServiceReconciliationFailed
+	ReceiverDeploymentReconciliationFailed
+
 	// Kafka Topic Reconciliation
 	KafkaTopicReconciliationFailed
 
@@ -56,10 +56,10 @@ func (et CoreV1EventType) String() string {
 		eventTypeString = "KafkaChannelServiceReconciliationFailed"
 	case ChannelUpdateFailed:
 		eventTypeString = "ChannelUpdateFailed"
-	case ChannelServiceReconciliationFailed:
-		eventTypeString = "ChannelServiceReconciliationFailed"
-	case ChannelDeploymentReconciliationFailed:
-		eventTypeString = "ChannelDeploymentReconciliationFailed"
+	case ReceiverServiceReconciliationFailed:
+		eventTypeString = "ReceiverServiceReconciliationFailed"
+	case ReceiverDeploymentReconciliationFailed:
+		eventTypeString = "ReceiverDeploymentReconciliationFailed"
 	case ChannelStatusReconciliationFailed:
 		eventTypeString = "ChannelStatusReconciliationFailed"
 	case KafkaTopicReconciliationFailed:
diff --git a/pkg/channel/distributed/controller/kafkachannel/channel.go b/pkg/channel/distributed/controller/kafkachannel/channel.go
index cc34dcd24..2ca8e82b0 100644
--- a/pkg/channel/distributed/controller/kafkachannel/channel.go
+++ b/pkg/channel/distributed/controller/kafkachannel/channel.go
@@ -106,8 +106,8 @@ func (r *Reconciler) newKafkaChannelService(channel *kafkav1beta1.KafkaChannel)
 	// Get The KafkaChannel Service Name
 	serviceName := kafkautil.AppendKafkaChannelServiceNameSuffix(channel.Name)
 
-	// Get The Dispatcher Service Name For The Channel (One Channel Service Per KafkaChannel Instance)
-	deploymentName := util.ChannelDnsSafeName(r.kafkaSecretName(channel))
+	// Get The Receiver Service Name For The Kafka Secret (One Receiver Service Per Kafka Secret)
+	deploymentName := util.ReceiverDnsSafeName(r.kafkaSecretName(channel))
 	serviceAddress := fmt.Sprintf("%s.%s.svc.%s", deploymentName, commonconstants.KnativeEventingNamespace, network.GetClusterDomainName())
 
 	// Create & Return The Service Model
@@ -120,7 +120,7 @@ func (r *Reconciler) newKafkaChannelService(channel *kafkav1beta1.KafkaChannel)
 			Name:      serviceName,       // Must Match KafkaChannel For HOST Parsing In Channel Implementation!
 			Namespace: channel.Namespace, // Must Match KafkaChannel For HOST Parsing In Channel Implementation!
 			Labels: map[string]string{
-				constants.KafkaChannelChannelLabel:   "true",                               // Identifies the Service as being a KafkaChannel "Channel"
+				constants.KafkaChannelReceiverLabel:  "true",                               // Identifies the Service as being a KafkaChannel "Channel"
 				constants.KafkaChannelNameLabel:      channel.Name,                         // Identifies the Service's Owning KafkaChannel's Name
 				constants.KafkaChannelNamespaceLabel: channel.Namespace,                    // Identifies the Service's Owning KafkaChannel's Namespace
 				constants.K8sAppChannelSelectorLabel: constants.K8sAppChannelSelectorValue, // Prometheus ServiceMonitor
diff --git a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
index 85748d870..a8a2225ce 100644
--- a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
+++ b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
@@ -387,6 +387,6 @@ func (r *Reconciler) dispatcherDeploymentEnvVars(channel *kafkav1beta1.KafkaChan
 		})
 	}
 
-	// Return The Channel Deployment EnvVars Array
+	// Return The Dispatcher Deployment EnvVars Array
 	return envVars, nil
 }
diff --git a/pkg/channel/distributed/controller/kafkasecret/channel.go b/pkg/channel/distributed/controller/kafkasecret/receiver.go
similarity index 60%
rename from pkg/channel/distributed/controller/kafkasecret/channel.go
rename to pkg/channel/distributed/controller/kafkasecret/receiver.go
index a85e290fa..31675aa25 100644
--- a/pkg/channel/distributed/controller/kafkasecret/channel.go
+++ b/pkg/channel/distributed/controller/kafkasecret/receiver.go
@@ -22,35 +22,35 @@ import (
 	"knative.dev/pkg/system"
 )
 
-// Reconcile The "Channel" Inbound For The Specified Kafka Secret
+// Reconcile The Receiver (Kafka Producer) For The Specified KafkaChannel
 func (r *Reconciler) reconcileChannel(ctx context.Context, secret *corev1.Secret) error {
 
 	// Get Secret Specific Logger
 	logger := util.SecretLogger(r.logger, secret)
 
-	// Reconcile The Channel's Service
-	serviceErr := r.reconcileChannelService(ctx, secret)
+	// Reconcile The Receiver Service
+	serviceErr := r.reconcileReceiverService(ctx, secret)
 	if serviceErr != nil {
-		controller.GetEventRecorder(ctx).Eventf(secret, corev1.EventTypeWarning, event.ChannelServiceReconciliationFailed.String(), "Failed To Reconcile Channel Service: %v", serviceErr)
-		logger.Error("Failed To Reconcile Channel Service", zap.Error(serviceErr))
+		controller.GetEventRecorder(ctx).Eventf(secret, corev1.EventTypeWarning, event.ReceiverServiceReconciliationFailed.String(), "Failed To Reconcile Receiver Service: %v", serviceErr)
+		logger.Error("Failed To Reconcile Receiver Service", zap.Error(serviceErr))
 	} else {
-		logger.Info("Successfully Reconciled Channel Service")
+		logger.Info("Successfully Reconciled Receiver Service")
 	}
 
-	// Reconcile The Channel's Deployment
-	deploymentErr := r.reconcileChannelDeployment(ctx, secret)
+	// Reconcile The Receiver Deployment
+	deploymentErr := r.reconcileReceiverDeployment(ctx, secret)
 	if deploymentErr != nil {
-		controller.GetEventRecorder(ctx).Eventf(secret, corev1.EventTypeWarning, event.ChannelDeploymentReconciliationFailed.String(), "Failed To Reconcile Channel Deployment: %v", deploymentErr)
-		logger.Error("Failed To Reconcile Channel Deployment", zap.Error(deploymentErr))
+		controller.GetEventRecorder(ctx).Eventf(secret, corev1.EventTypeWarning, event.ReceiverDeploymentReconciliationFailed.String(), "Failed To Reconcile Receiver Deployment: %v", deploymentErr)
+		logger.Error("Failed To Reconcile Receiver Deployment", zap.Error(deploymentErr))
 	} else {
-		logger.Info("Successfully Reconciled Channel Deployment")
+		logger.Info("Successfully Reconciled Receiver Deployment")
 	}
 
 	// Reconcile Channel's KafkaChannel Status
 	statusErr := r.reconcileKafkaChannelStatus(ctx,
 		secret,
-		serviceErr == nil, event.ChannelServiceReconciliationFailed.String(), fmt.Sprintf("Channel Service Failed: %v", serviceErr),
-		deploymentErr == nil, event.ChannelDeploymentReconciliationFailed.String(), fmt.Sprintf("Channel Deployment Failed: %v", deploymentErr))
+		serviceErr == nil, event.ReceiverServiceReconciliationFailed.String(), fmt.Sprintf("Receiver Service Failed: %v", serviceErr),
+		deploymentErr == nil, event.ReceiverDeploymentReconciliationFailed.String(), fmt.Sprintf("Receiver Deployment Failed: %v", deploymentErr))
 	if statusErr != nil {
 		controller.GetEventRecorder(ctx).Eventf(secret, corev1.EventTypeWarning, event.ChannelStatusReconciliationFailed.String(), "Failed To Reconcile Channel's KafkaChannel Status: %v", statusErr)
 		logger.Error("Failed To Reconcile KafkaChannel Status", zap.Error(statusErr))
@@ -67,65 +67,65 @@ func (r *Reconciler) reconcileChannel(ctx context.Context, secret *corev1.Secret
 }
 
 //
-// Kafka Channel Service
+// Kafka Receiver Service
 //
 
-// Reconcile The Kafka Channel Service
-func (r *Reconciler) reconcileChannelService(ctx context.Context, secret *corev1.Secret) error {
+// Reconcile The Receiver Service
+func (r *Reconciler) reconcileReceiverService(ctx context.Context, secret *corev1.Secret) error {
 
-	// Attempt To Get The Service Associated With The Specified Secret
-	_, err := r.getChannelService(secret)
+	// Attempt To Get The Receiver Service Associated With The Specified Secret
+	_, err := r.getReceiverService(secret)
 	if err != nil {
 
 		// If The Service Was Not Found - Then Create A New One For The Secret
 		if errors.IsNotFound(err) {
 
-			// Then Create The New Kafka Channel Service
-			r.logger.Info("Channel Service Not Found - Creating New One")
-			service := r.newChannelService(secret)
+			// Then Create The New Receiver Service
+			r.logger.Info("Receiver Service Not Found - Creating New One")
+			service := r.newReceiverService(secret)
 			_, err = r.kubeClientset.CoreV1().Services(service.Namespace).Create(ctx, service, metav1.CreateOptions{})
 			if err != nil {
-				r.logger.Error("Failed To Create Channel Service", zap.Error(err))
+				r.logger.Error("Failed To Create Receiver Service", zap.Error(err))
 				return err
 			} else {
-				r.logger.Info("Successfully Created Channel Service")
+				r.logger.Info("Successfully Created Receiver Service")
 				return nil
 			}
 
 		} else {
 
-			// Failed In Attempt To Get Kafka Channel Service From K8S
-			r.logger.Error("Failed To Get Channel Service", zap.Error(err))
+			// Failed In Attempt To Get Receiver Service From K8S
+			r.logger.Error("Failed To Get Receiver Service", zap.Error(err))
 			return err
 		}
 	} else {
 
-		// Verified The Channel Service Exists
-		r.logger.Info("Successfully Verified Channel Service")
+		// Verified The Receiver Service Exists
+		r.logger.Info("Successfully Verified Receiver Service")
 		return nil
 	}
 }
 
-// Get The Kafka Service Associated With The Specified Channel
-func (r *Reconciler) getChannelService(secret *corev1.Secret) (*corev1.Service, error) {
+// Get The Kafka Receiver Service Associated With The Specified Channel
+func (r *Reconciler) getReceiverService(secret *corev1.Secret) (*corev1.Service, error) {
 
-	// Get The Dispatcher Deployment Name For The Channel - Use Same For Service
-	deploymentName := util.ChannelDnsSafeName(secret.Name)
+	// Get The Receiver Deployment Name For The Receiver - Use Same For Service
+	deploymentName := util.ReceiverDnsSafeName(secret.Name)
 
-	// Get The Service By Namespace / Name
+	// Get The Receiver Service By Namespace / Name
 	service, err := r.serviceLister.Services(commonconstants.KnativeEventingNamespace).Get(deploymentName)
 
 	// Return The Results
 	return service, err
 }
 
-// Create Kafka Service Model For The Specified Channel
-func (r *Reconciler) newChannelService(secret *corev1.Secret) *corev1.Service {
+// Create Receiver Service Model For The Specified Secret
+func (r *Reconciler) newReceiverService(secret *corev1.Secret) *corev1.Service {
 
-	// Get The Dispatcher Deployment Name For The Channel - Use Same For Service
-	deploymentName := util.ChannelDnsSafeName(secret.Name)
+	// Get The Receiver Deployment Name For The Secret - Use Same For Service
+	deploymentName := util.ReceiverDnsSafeName(secret.Name)
 
-	// Create & Return The Service Model
+	// Create & Return The Receiver Service Model
 	return &corev1.Service{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: corev1.SchemeGroupVersion.String(),
@@ -135,7 +135,7 @@ func (r *Reconciler) newChannelService(secret *corev1.Secret) *corev1.Service {
 			Name:      deploymentName,
 			Namespace: commonconstants.KnativeEventingNamespace,
 			Labels: map[string]string{
-				constants.KafkaChannelChannelLabel:   "true",                               // Allows for identification of KafkaChannels
+				constants.KafkaChannelReceiverLabel:  "true",                               // Allows for identification of Receivers
 				constants.K8sAppChannelSelectorLabel: constants.K8sAppChannelSelectorValue, // Prometheus ServiceMonitor
 			},
 			OwnerReferences: []metav1.OwnerReference{
@@ -163,80 +163,80 @@ func (r *Reconciler) newChannelService(secret *corev1.Secret) *corev1.Service {
 }
 
 //
-// KafkaChannel Deployment - The Kafka Producer Implementation
+// Kafka Receiver Deployment - The Kafka Producer Implementation
 //
 
-// Reconcile The Kafka Channel Deployment
-func (r *Reconciler) reconcileChannelDeployment(ctx context.Context, secret *corev1.Secret) error {
+// Reconcile The Receiver Deployment
+func (r *Reconciler) reconcileReceiverDeployment(ctx context.Context, secret *corev1.Secret) error {
 
-	// Attempt To Get The KafkaChannel Deployment Associated With The Specified Channel
-	_, err := r.getChannelDeployment(secret)
+	// Attempt To Get The Receiver Deployment Associated With The Specified Secret
+	_, err := r.getReceiverDeployment(secret)
 	if err != nil {
 
-		// If The KafkaChannel Deployment Was Not Found - Then Create A New Deployment For The Channel
+		// If The Receiver Deployment Was Not Found - Then Create A New Deployment For The Secret
 		if errors.IsNotFound(err) {
 
-			// Then Create The New Deployment
-			r.logger.Info("KafkaChannel Deployment Not Found - Creating New One")
+			// Then Create The New Receiver Deployment
+			r.logger.Info("Receiver Deployment Not Found - Creating New One")
 			deployment, err := r.newChannelDeployment(secret)
 			if err != nil {
-				r.logger.Error("Failed To Create KafkaChannel Deployment YAML", zap.Error(err))
+				r.logger.Error("Failed To Create Receiver Deployment YAML", zap.Error(err))
 				return err
 			} else {
 				_, err = r.kubeClientset.AppsV1().Deployments(deployment.Namespace).Create(ctx, deployment, metav1.CreateOptions{})
 				if err != nil {
-					r.logger.Error("Failed To Create KafkaChannel Deployment", zap.Error(err))
+					r.logger.Error("Failed To Create Receiver Deployment", zap.Error(err))
 					return err
 				} else {
-					r.logger.Info("Successfully Created KafkaChannel Deployment")
+					r.logger.Info("Successfully Created Receiver Deployment")
 					return nil
 				}
 			}
 
 		} else {
 
-			// Failed In Attempt To Get Deployment From K8S
-			r.logger.Error("Failed To Get KafkaChannel Deployment", zap.Error(err))
+			// Failed In Attempt To Get Receiver Deployment From K8S
+			r.logger.Error("Failed To Get Receiver Deployment", zap.Error(err))
 			return err
 		}
 	} else {
 
-		// Verified The Channel Service Exists
-		r.logger.Info("Successfully Verified Channel Deployment")
+		// Verified The Receiver Deployment Exists
+		r.logger.Info("Successfully Verified Receiver Deployment")
 		return nil
 	}
 }
 
-// Get The Kafka Channel Deployment Associated With The Specified Secret
-func (r *Reconciler) getChannelDeployment(secret *corev1.Secret) (*appsv1.Deployment, error) {
+// Get The Receiver Deployment Associated With The Specified Secret
+func (r *Reconciler) getReceiverDeployment(secret *corev1.Secret) (*appsv1.Deployment, error) {
 
-	// Get The Channel Deployment Name (One Channel Deployment Per Kafka Auth Secret)
-	deploymentName := util.ChannelDnsSafeName(secret.Name)
+	// Get The Receiver Deployment Name (One Receiver Deployment Per Kafka Auth Secret)
+	deploymentName := util.ReceiverDnsSafeName(secret.Name)
 
-	// Get The Channel Deployment By Namespace / Name
+	// Get The Receiver Deployment By Namespace / Name
 	deployment, err := r.deploymentLister.Deployments(commonconstants.KnativeEventingNamespace).Get(deploymentName)
 
 	// Return The Results
 	return deployment, err
 }
 
-// Create Kafka Channel Deployment Model For The Specified Secret
+// Create Receiver Deployment Model For The Specified Secret
 func (r *Reconciler) newChannelDeployment(secret *corev1.Secret) (*appsv1.Deployment, error) {
 
-	// Get The Channel Deployment Name (One Channel Deployment Per Kafka Auth Secret)
-	deploymentName := util.ChannelDnsSafeName(secret.Name)
+	// Get The Receiver Deployment Name (One Receiver Deployment Per Kafka Auth Secret)
+	deploymentName := util.ReceiverDnsSafeName(secret.Name)
 
 	// Replicas Int Value For De-Referencing
-	replicas := int32(r.config.Channel.Replicas)
+	replicas := int32(r.config.Receiver.Replicas)
 
-	// Create The Channel Container Environment Variables
-	channelEnvVars, err := r.channelDeploymentEnvVars(secret)
+	// Create The Receiver Container Environment Variables
+	channelEnvVars, err := r.receiverDeploymentEnvVars(secret)
 	if err != nil {
-		r.logger.Error("Failed To Create Channel Deployment Environment Variables", zap.Error(err))
+		r.logger.Error("Failed To Create Receiver Deployment Environment Variables", zap.Error(err))
 		return nil, err
 	}
 
-	// Create & Return The Channel's Deployment
+	// Create The Receiver Deployment
 	deployment := &appsv1.Deployment{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: appsv1.SchemeGroupVersion.String(),
@@ -246,8 +246,8 @@ func (r *Reconciler) newChannelDeployment(secret *corev1.Secret) (*appsv1.Deploy
 			Name:      deploymentName,
 			Namespace: commonconstants.KnativeEventingNamespace,
 			Labels: map[string]string{
-				constants.AppLabel:                 deploymentName, // Matches Service Selector Key/Value Below
-				constants.KafkaChannelChannelLabel: "true",         // Allows for identification of KafkaChannels
+				constants.AppLabel:                  deploymentName, // Matches Service Selector Key/Value Below
+				constants.KafkaChannelReceiverLabel: "true",         // Allows for identification of Receivers
 			},
 			OwnerReferences: []metav1.OwnerReference{
 				util.NewSecretOwnerReference(secret),
@@ -291,7 +291,7 @@ func (r *Reconciler) newChannelDeployment(secret *corev1.Secret) (*appsv1.Deploy
 								InitialDelaySeconds: constants.ChannelReadinessDelay,
 								PeriodSeconds:       constants.ChannelReadinessPeriod,
 							},
-							Image: r.environment.ChannelImage,
+							Image: r.environment.ReceiverImage,
 							Ports: []corev1.ContainerPort{
 								{
 									Name:          "server",
@@ -302,12 +302,12 @@ func (r *Reconciler) newChannelDeployment(secret *corev1.Secret) (*appsv1.Deploy
 							ImagePullPolicy: corev1.PullIfNotPresent,
 							Resources: corev1.ResourceRequirements{
 								Requests: corev1.ResourceList{
-									corev1.ResourceCPU:    r.config.Channel.CpuRequest,
-									corev1.ResourceMemory: r.config.Channel.MemoryRequest,
+									corev1.ResourceCPU:    r.config.Receiver.CpuRequest,
+									corev1.ResourceMemory: r.config.Receiver.MemoryRequest,
 								},
 								Limits: corev1.ResourceList{
-									corev1.ResourceCPU:    r.config.Channel.CpuLimit,
-									corev1.ResourceMemory: r.config.Channel.MemoryLimit,
+									corev1.ResourceCPU:    r.config.Receiver.CpuLimit,
+									corev1.ResourceMemory: r.config.Receiver.MemoryLimit,
 								},
 							},
 						},
@@ -317,14 +317,14 @@ func (r *Reconciler) newChannelDeployment(secret *corev1.Secret) (*appsv1.Deploy
 		},
 	}
 
-	// Return Channel Deployment
+	// Return Receiver Deployment
 	return deployment, nil
 }
 
-// Create The Kafka Channel Deployment's Env Vars
-func (r *Reconciler) channelDeploymentEnvVars(secret *corev1.Secret) ([]corev1.EnvVar, error) {
+// Create The Receiver Deployment's Env Vars
+func (r *Reconciler) receiverDeploymentEnvVars(secret *corev1.Secret) ([]corev1.EnvVar, error) {
 
-	// Create The Channel Deployment EnvVars
+	// Create The Receiver Deployment EnvVars
 	envVars := []corev1.EnvVar{
 		{
 			Name:  system.NamespaceEnvKey,
@@ -336,7 +336,7 @@ func (r *Reconciler) channelDeploymentEnvVars(secret *corev1.Secret) ([]corev1.E
 		},
 		{
 			Name:  commonenv.ServiceNameEnvVarKey,
-			Value: util.ChannelDnsSafeName(secret.Name),
+			Value: util.ReceiverDnsSafeName(secret.Name),
 		},
 		{
 			Name:  commonenv.MetricsPortEnvVarKey,
@@ -385,6 +385,6 @@ func (r *Reconciler) channelDeploymentEnvVars(secret *corev1.Secret) ([]corev1.E
 		},
 	})
 
-	// Return The Channel Deployment EnvVars Array
+	// Return The Receiver Deployment EnvVars Array
 	return envVars, nil
 }
diff --git a/pkg/channel/distributed/controller/kafkasecret/reconciler.go b/pkg/channel/distributed/controller/kafkasecret/reconciler.go
index c9a73c558..cb8d133d8 100644
--- a/pkg/channel/distributed/controller/kafkasecret/reconciler.go
+++ b/pkg/channel/distributed/controller/kafkasecret/reconciler.go
@@ -52,7 +52,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, secret *corev1.Secret) r
 	}
 
 	// Return Success
-	logger.Info("Successfully Reconciled Kafka Secret", zap.Any("Channel", secret.Name))
+	logger.Info("Successfully Reconciled Kafka Secret", zap.Any("Secret", secret.Name))
 	return reconciler.NewEvent(corev1.EventTypeNormal, event.KafkaSecretReconciled.String(), "Kafka Secret Reconciled Successfully: \"%s/%s\"", secret.Namespace, secret.Name)
 }
 
@@ -63,7 +63,7 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, secret *corev1.Secret) re
 	r.logger.Debug("<==========  START KAFKA-SECRET FINALIZATION  ==========>")
 	logger := r.logger.With(zap.String("Secret", secret.Name))
 
-	// Reconcile The Affected KafkaChannel Status To Indicate The Channel Service/Deployment Are Not Longer Available
+	// Reconcile The Affected KafkaChannel Status To Indicate The Receiver Service/Deployment Is No Longer Available
 	err := r.reconcileKafkaChannelStatus(ctx,
 		secret,
 		false, "ChannelServiceUnavailable", "Kafka Auth Secret Finalized",
diff --git a/pkg/channel/distributed/controller/util/channel.go b/pkg/channel/distributed/controller/util/channel.go
index a469dafe1..96cdedff0 100644
--- a/pkg/channel/distributed/controller/util/channel.go
+++ b/pkg/channel/distributed/controller/util/channel.go
@@ -44,19 +44,19 @@ func NewChannelOwnerReference(channel *kafkav1beta1.KafkaChannel) metav1.OwnerRe
 }
 
 //
-// Create A DNS Safe Name For The Channel Deployment Using The Specified Kafka Secret
+// Create A DNS Safe Name For The Receiver Deployment Using The Specified Kafka Secret
 //
-// Note - The current implementation creates a single Channel Deployment for each
+// Note - The current implementation creates a single Receiver Deployment for each
 //        Kafka Authentication (K8S Secrets) instance.
 //
-func ChannelDnsSafeName(kafkaSecretName string) string {
+func ReceiverDnsSafeName(kafkaSecretName string) string {
 
 	// In order for the resulting name to be a valid DNS component it's length must be no more than 63 characters.
-	// We are consuming 17 chars for the component separators, hash, and Channel suffix, which reduces the
-	// available length to 46. We will allocate 41 characters to the kafka secret name leaving an extra buffer.
-	safeSecretName := GenerateValidDnsName(kafkaSecretName, 41, true, false)
+	// We are consuming 18 chars for the component separators, hash, and Receiver suffix, which reduces the
+	// available length to 45. We will allocate 40 characters to the kafka secret name leaving an extra buffer.
+	safeSecretName := GenerateValidDnsName(kafkaSecretName, 40, true, false)
 
-	return fmt.Sprintf("%s-%s-channel", safeSecretName, GenerateHash(kafkaSecretName, 8))
+	return fmt.Sprintf("%s-%s-receiver", safeSecretName, GenerateHash(kafkaSecretName, 8))
 }
 
 // Channel Host Naming Utility
diff --git a/pkg/channel/distributed/dispatcher/constants/constants.go b/pkg/channel/distributed/dispatcher/constants/constants.go
index f0d0fce04..352ce067f 100644
--- a/pkg/channel/distributed/dispatcher/constants/constants.go
+++ b/pkg/channel/distributed/dispatcher/constants/constants.go
@@ -2,8 +2,5 @@ package constants
 
 // Global Constants
 const (
-	// Default Values If Optional config-eventing-kafka ConfigMap Defaults Not Specified
-	DefaultEventRetryInitialIntervalMillis = 500    // 0.5 seconds
-	DefaultEventRetryTimeMillisMax         = 300000 // 5 minutes
-	DefaultExponentialBackoff              = true   // Enabled
+	Component = "eventing-kafka-channel-dispatcher"
 )
diff --git a/pkg/channel/distributed/receiver/README.md b/pkg/channel/distributed/receiver/README.md
index 093b2d507..a2d59295d 100644
--- a/pkg/channel/distributed/receiver/README.md
+++ b/pkg/channel/distributed/receiver/README.md
@@ -1,10 +1,10 @@
-# Eventing-Kafka Channel
+# Eventing-Kafka Receiver
 
-The "Channel" implementation is a Kafka Producer which is responsible for
+The Receiver implementation is a Kafka Producer which is responsible for
 receiving CloudEvents, converting them to Kafka Messages, and writing them
 to the appropriate Kafka Topic.
 
-A unique Channel Deployment / Service is created for every Kafka Secret (in
+A unique Deployment / Service is created for every Kafka Secret (in
 the knative-eventing namespace and labelled with
 `eventing-kafka.knative.dev/kafka-secret: "true"`).  The single Deployment
 is horizontally scalable as necessary (via controller environment variables.) This
@@ -14,22 +14,23 @@ volume and multi-tenant use cases.
 An additional Service for each KafkaChannel is created in the user namespace
 where the KafkaChannel exists.  This is the actual endpoint of the `KafkaChannel`
 This service forwards requests to the service mentioned above in the
-knative-eventing namespace.  This is all necessary so that the Channel
+knative-eventing namespace.  This is all necessary so that the Receiver
 implementation can use the hostname of the request to map incoming CloudEvents
 to the appropriate Kafka Topic.
 
-The Kafka brokers and credentials are obtained from mounted Secret data from the aforementiond Kafka Secret.
+The Kafka brokers and credentials are obtained from mounted Secret data from
+the aforementiond Kafka Secret.
 
 ## Tracing, Profiling, and Metrics
 
-The Channel makes use of the infrastructure surrounding the config-tracing and config-observability
+The Receiver makes use of the infrastructure surrounding the config-tracing and config-observability
 configmaps (see [Accessing CloudEvent traces](https://knative.dev/docs/eventing/accessing-traces) and
 [Installing logging, metrics, and traces](https://knative.dev/docs/serving/installing-logging-metrics-traces)
 for more information on the basic Knative-Eventing concepts behind these features.  The default behavior for
 tracing and profiling is provided such that accessing a tracing server (such as Zipkin), and the debug profiling
 information, should work as described in those links.  For example, you might access your zipkin server
 via `http://localhost:8001/api/v1/namespaces/knative-eventing/services/zipkin:9411/proxy/zipkin` after running a
-"kubectl proxy" command or your profiling server via `http://localhost:8008/debug/pprof` after executing
+"kubectl proxy" command, or the profiling server via `http://localhost:8008/debug/pprof` after executing
 "kubectl -n knative-eventing port-forward my-channel-pod-name 8008:8008"
 
 Eventing-Kafka does provide some of its own custom metrics that use the Prometheus server provided by
diff --git a/pkg/channel/distributed/receiver/constants/constants.go b/pkg/channel/distributed/receiver/constants/constants.go
index 9b2326bd2..0bfd807fd 100644
--- a/pkg/channel/distributed/receiver/constants/constants.go
+++ b/pkg/channel/distributed/receiver/constants/constants.go
@@ -4,7 +4,7 @@ import "time"
 
 // Global Constants
 const (
-	Component = "KafkaChannel"
+	Component = "eventing-kafka-channel-receiver"
 
 	MetricsInterval = 5 * time.Second
 
diff --git a/pkg/channel/distributed/receiver/env/environment.go b/pkg/channel/distributed/receiver/env/environment.go
index 3a4c5082c..7efce9575 100644
--- a/pkg/channel/distributed/receiver/env/environment.go
+++ b/pkg/channel/distributed/receiver/env/environment.go
@@ -75,9 +75,9 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 		safeEnvironment.KafkaPassword = "*************"
 	}
 
-	// Log The Channel Configuration Loaded From Environment Variables
+	// Log The Receiver Configuration Loaded From Environment Variables
 	logger.Info("Environment Variables", zap.Any("Environment", safeEnvironment))
 
-	// Return The Populated Channel Configuration Environment Structure
+	// Return The Populated Environment Structure
 	return environment, nil
 }


Test patch:
diff --git a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
index 870cb8c23..a633319a5 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
@@ -24,7 +24,7 @@ import (
 const (
 	// EKDefaultConfigYaml is intended to match what's in 200-eventing-kafka-configmap.yaml
 	EKDefaultConfigYaml = `
-channel:
+receiver:
   cpuLimit: 200m
   cpuRequest: 100m
   memoryLimit: 100Mi
@@ -195,11 +195,11 @@ func TestLoadDefaultSaramaSettings(t *testing.T) {
 
 	// Make sure all of our default eventing-kafka settings were loaded properly
 	// Specifically checking the type (e.g. int64, int16, int) is important
-	assert.Equal(t, resource.MustParse("200m"), configuration.Channel.CpuLimit)
-	assert.Equal(t, resource.MustParse("100m"), configuration.Channel.CpuRequest)
-	assert.Equal(t, resource.MustParse("100Mi"), configuration.Channel.MemoryLimit)
-	assert.Equal(t, resource.MustParse("50Mi"), configuration.Channel.MemoryRequest)
-	assert.Equal(t, 1, configuration.Channel.Replicas)
+	assert.Equal(t, resource.MustParse("200m"), configuration.Receiver.CpuLimit)
+	assert.Equal(t, resource.MustParse("100m"), configuration.Receiver.CpuRequest)
+	assert.Equal(t, resource.MustParse("100Mi"), configuration.Receiver.MemoryLimit)
+	assert.Equal(t, resource.MustParse("50Mi"), configuration.Receiver.MemoryRequest)
+	assert.Equal(t, 1, configuration.Receiver.Replicas)
 	assert.Equal(t, int32(4), configuration.Kafka.Topic.DefaultNumPartitions)
 	assert.Equal(t, int16(1), configuration.Kafka.Topic.DefaultReplicationFactor)
 	assert.Equal(t, int64(604800000), configuration.Kafka.Topic.DefaultRetentionMillis)
diff --git a/pkg/channel/distributed/controller/config/config_test.go b/pkg/channel/distributed/controller/config/config_test.go
index d3631f15f..cd1a196c7 100644
--- a/pkg/channel/distributed/controller/config/config_test.go
+++ b/pkg/channel/distributed/controller/config/config_test.go
@@ -123,29 +123,29 @@ func TestVerifyConfiguration(t *testing.T) {
 	testCase.expectedError = ControllerConfigurationError("Dispatcher.Replicas must be > 0")
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Channel.CpuLimit")
+	testCase = getValidTestCase("Invalid Config - Receiver.CpuLimit")
 	testCase.channelCpuLimit = resource.Quantity{}
-	testCase.expectedError = ControllerConfigurationError("Channel.CpuLimit must be nonzero")
+	testCase.expectedError = ControllerConfigurationError("Receiver.CpuLimit must be nonzero")
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Channel.CpuRequest")
+	testCase = getValidTestCase("Invalid Config - Receiver.CpuRequest")
 	testCase.channelCpuRequest = resource.Quantity{}
-	testCase.expectedError = ControllerConfigurationError("Channel.CpuRequest must be nonzero")
+	testCase.expectedError = ControllerConfigurationError("Receiver.CpuRequest must be nonzero")
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Channel.MemoryLimit")
+	testCase = getValidTestCase("Invalid Config - Receiver.MemoryLimit")
 	testCase.channelMemoryLimit = resource.Quantity{}
-	testCase.expectedError = ControllerConfigurationError("Channel.MemoryLimit must be nonzero")
+	testCase.expectedError = ControllerConfigurationError("Receiver.MemoryLimit must be nonzero")
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Channel.MemoryRequest")
+	testCase = getValidTestCase("Invalid Config - Receiver.MemoryRequest")
 	testCase.channelMemoryRequest = resource.Quantity{}
-	testCase.expectedError = ControllerConfigurationError("Channel.MemoryRequest must be nonzero")
+	testCase.expectedError = ControllerConfigurationError("Receiver.MemoryRequest must be nonzero")
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Channel.Replicas")
+	testCase = getValidTestCase("Invalid Config - Receiver.Replicas")
 	testCase.channelReplicas = -1
-	testCase.expectedError = ControllerConfigurationError("Channel.Replicas must be > 0")
+	testCase.expectedError = ControllerConfigurationError("Receiver.Replicas must be > 0")
 	testCases = append(testCases, testCase)
 
 	testCase = getValidTestCase("Invalid Config - Kafka.Provider")
@@ -166,11 +166,11 @@ func TestVerifyConfiguration(t *testing.T) {
 		testConfig.Dispatcher.MemoryLimit = testCase.dispatcherMemoryLimit
 		testConfig.Dispatcher.MemoryRequest = testCase.dispatcherMemoryRequest
 		testConfig.Dispatcher.Replicas = testCase.dispatcherReplicas
-		testConfig.Channel.CpuLimit = testCase.channelCpuLimit
-		testConfig.Channel.CpuRequest = testCase.channelCpuRequest
-		testConfig.Channel.MemoryLimit = testCase.channelMemoryLimit
-		testConfig.Channel.MemoryRequest = testCase.channelMemoryRequest
-		testConfig.Channel.Replicas = testCase.channelReplicas
+		testConfig.Receiver.CpuLimit = testCase.channelCpuLimit
+		testConfig.Receiver.CpuRequest = testCase.channelCpuRequest
+		testConfig.Receiver.MemoryLimit = testCase.channelMemoryLimit
+		testConfig.Receiver.MemoryRequest = testCase.channelMemoryRequest
+		testConfig.Receiver.Replicas = testCase.channelReplicas
 
 		// Perform The Test
 		err := VerifyConfiguration(testConfig)
@@ -187,11 +187,11 @@ func TestVerifyConfiguration(t *testing.T) {
 			assert.Equal(t, testCase.dispatcherMemoryLimit, testConfig.Dispatcher.MemoryLimit)
 			assert.Equal(t, testCase.dispatcherMemoryRequest, testConfig.Dispatcher.MemoryRequest)
 			assert.Equal(t, testCase.dispatcherReplicas, testConfig.Dispatcher.Replicas)
-			assert.Equal(t, testCase.channelCpuLimit, testConfig.Channel.CpuLimit)
-			assert.Equal(t, testCase.channelCpuRequest, testConfig.Channel.CpuRequest)
-			assert.Equal(t, testCase.channelMemoryLimit, testConfig.Channel.MemoryLimit)
-			assert.Equal(t, testCase.channelMemoryRequest, testConfig.Channel.MemoryRequest)
-			assert.Equal(t, testCase.channelReplicas, testConfig.Channel.Replicas)
+			assert.Equal(t, testCase.channelCpuLimit, testConfig.Receiver.CpuLimit)
+			assert.Equal(t, testCase.channelCpuRequest, testConfig.Receiver.CpuRequest)
+			assert.Equal(t, testCase.channelMemoryLimit, testConfig.Receiver.MemoryLimit)
+			assert.Equal(t, testCase.channelMemoryRequest, testConfig.Receiver.MemoryRequest)
+			assert.Equal(t, testCase.channelReplicas, testConfig.Receiver.Replicas)
 		} else {
 			assert.Equal(t, testCase.expectedError, err)
 		}
diff --git a/pkg/channel/distributed/controller/env/environment_test.go b/pkg/channel/distributed/controller/env/environment_test.go
index 510e5464d..49e5af848 100644
--- a/pkg/channel/distributed/controller/env/environment_test.go
+++ b/pkg/channel/distributed/controller/env/environment_test.go
@@ -22,7 +22,7 @@ const (
 
 	dispatcherImage = "TestDispatcherImage"
 
-	channelImage = "TestChannelImage"
+	receiverImage = "TestReceiverImage"
 )
 
 // Define The TestCase Struct
@@ -73,9 +73,9 @@ func TestGetEnvironment(t *testing.T) {
 	testCase.expectedError = getMissingRequiredEnvironmentVariableError(DispatcherImageEnvVarKey)
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Missing Required Config - ChannelImage")
+	testCase = getValidTestCase("Missing Required Config - ReceiverImage")
 	testCase.channelImage = ""
-	testCase.expectedError = getMissingRequiredEnvironmentVariableError(ChannelImageEnvVarKey)
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(ReceiverImageEnvVarKey)
 	testCases = append(testCases, testCase)
 
 	// Loop Over All The TestCases
@@ -87,7 +87,7 @@ func TestGetEnvironment(t *testing.T) {
 		assertSetenv(t, env.MetricsDomainEnvVarKey, testCase.metricsDomain)
 		assertSetenvNonempty(t, env.MetricsPortEnvVarKey, testCase.metricsPort)
 		assertSetenv(t, DispatcherImageEnvVarKey, testCase.dispatcherImage)
-		assertSetenv(t, ChannelImageEnvVarKey, testCase.channelImage)
+		assertSetenv(t, ReceiverImageEnvVarKey, testCase.channelImage)
 
 		// Perform The Test
 		environment, err := GetEnvironment(logger)
@@ -99,7 +99,7 @@ func TestGetEnvironment(t *testing.T) {
 			assert.NotNil(t, environment)
 			assert.Equal(t, testCase.serviceAccount, environment.ServiceAccount)
 			assert.Equal(t, testCase.metricsPort, strconv.Itoa(environment.MetricsPort))
-			assert.Equal(t, testCase.channelImage, environment.ChannelImage)
+			assert.Equal(t, testCase.channelImage, environment.ReceiverImage)
 			assert.Equal(t, testCase.dispatcherImage, environment.DispatcherImage)
 
 		} else {
@@ -119,7 +119,7 @@ func getValidTestCase(name string) TestCase {
 		metricsDomain:         metricsDomain,
 		defaultKafkaConsumers: defaultKafkaConsumers,
 		dispatcherImage:       dispatcherImage,
-		channelImage:          channelImage,
+		channelImage:          receiverImage,
 		expectedError:         nil,
 	}
 }
diff --git a/pkg/channel/distributed/controller/event/event_test.go b/pkg/channel/distributed/controller/event/event_test.go
index 233b39e5c..7d914e576 100644
--- a/pkg/channel/distributed/controller/event/event_test.go
+++ b/pkg/channel/distributed/controller/event/event_test.go
@@ -12,9 +12,9 @@ func TestEventTypes(t *testing.T) {
 	performEventTypeStringTest(t, ClusterChannelProvisionerUpdateStatusFailed, "ClusterChannelProvisionerUpdateStatusFailed")
 	performEventTypeStringTest(t, KafkaChannelServiceReconciliationFailed, "KafkaChannelServiceReconciliationFailed")
 	performEventTypeStringTest(t, ChannelUpdateFailed, "ChannelUpdateFailed")
-	performEventTypeStringTest(t, ChannelServiceReconciliationFailed, "ChannelServiceReconciliationFailed")
-	performEventTypeStringTest(t, ChannelServiceReconciliationFailed, "ChannelServiceReconciliationFailed")
-	performEventTypeStringTest(t, ChannelDeploymentReconciliationFailed, "ChannelDeploymentReconciliationFailed")
+	performEventTypeStringTest(t, ReceiverServiceReconciliationFailed, "ReceiverServiceReconciliationFailed")
+	performEventTypeStringTest(t, ReceiverServiceReconciliationFailed, "ReceiverServiceReconciliationFailed")
+	performEventTypeStringTest(t, ReceiverDeploymentReconciliationFailed, "ReceiverDeploymentReconciliationFailed")
 	performEventTypeStringTest(t, KafkaTopicReconciliationFailed, "KafkaTopicReconciliationFailed")
 	performEventTypeStringTest(t, DispatcherServiceReconciliationFailed, "DispatcherServiceReconciliationFailed")
 	performEventTypeStringTest(t, DispatcherDeploymentReconciliationFailed, "DispatcherDeploymentReconciliationFailed")
diff --git a/pkg/channel/distributed/controller/kafkachannel/controller_test.go b/pkg/channel/distributed/controller/kafkachannel/controller_test.go
index 644ea834c..7c9deda10 100644
--- a/pkg/channel/distributed/controller/kafkachannel/controller_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/controller_test.go
@@ -77,6 +77,6 @@ func populateEnvironmentVariables(t *testing.T) {
 	assert.Nil(t, os.Setenv(commonenv.ServiceAccountEnvVarKey, controllertesting.ServiceAccount))
 	assert.Nil(t, os.Setenv(commonenv.MetricsDomainEnvVarKey, controllertesting.MetricsDomain))
 	assert.Nil(t, os.Setenv(commonenv.MetricsPortEnvVarKey, strconv.Itoa(controllertesting.MetricsPort)))
-	assert.Nil(t, os.Setenv(controllerenv.DispatcherImageEnvVarKey, controllertesting.MetricsDomain))
-	assert.Nil(t, os.Setenv(controllerenv.ChannelImageEnvVarKey, controllertesting.MetricsDomain))
+	assert.Nil(t, os.Setenv(controllerenv.DispatcherImageEnvVarKey, controllertesting.DispatcherImage))
+	assert.Nil(t, os.Setenv(controllerenv.ReceiverImageEnvVarKey, controllertesting.ReceiverImage))
 }
diff --git a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
index 9a8f342a7..371b1b6e8 100644
--- a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
@@ -265,14 +265,14 @@ func TestReconcile(t *testing.T) {
 					controllertesting.WithAddress,
 					controllertesting.WithInitializedConditions,
 					controllertesting.WithKafkaChannelServiceReady,
-					controllertesting.WithChannelServiceReady,
-					controllertesting.WithChannelDeploymentReady,
+					controllertesting.WithReceiverServiceReady,
+					controllertesting.WithReceiverDeploymentReady,
 					controllertesting.WithDispatcherDeploymentReady,
 					controllertesting.WithTopicReady,
 				),
 				controllertesting.NewKafkaChannelService(),
-				controllertesting.NewKafkaChannelChannelService(),
-				controllertesting.NewKafkaChannelChannelDeployment(),
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
 				controllertesting.NewKafkaChannelDispatcherDeployment(),
 			},
 			WantCreates: []runtime.Object{controllertesting.NewKafkaChannelDispatcherService()},
@@ -289,14 +289,14 @@ func TestReconcile(t *testing.T) {
 					controllertesting.WithAddress,
 					controllertesting.WithInitializedConditions,
 					controllertesting.WithKafkaChannelServiceReady,
-					controllertesting.WithChannelServiceReady,
-					controllertesting.WithChannelDeploymentReady,
+					controllertesting.WithReceiverServiceReady,
+					controllertesting.WithReceiverDeploymentReady,
 					controllertesting.WithDispatcherDeploymentReady,
 					controllertesting.WithTopicReady,
 				),
 				controllertesting.NewKafkaChannelService(),
-				controllertesting.NewKafkaChannelChannelService(),
-				controllertesting.NewKafkaChannelChannelDeployment(),
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
 				controllertesting.NewKafkaChannelDispatcherDeployment(),
 			},
 			WithReactors:      []clientgotesting.ReactionFunc{InduceFailure("create", "services")},
@@ -326,14 +326,14 @@ func TestReconcile(t *testing.T) {
 					controllertesting.WithAddress,
 					controllertesting.WithInitializedConditions,
 					controllertesting.WithKafkaChannelServiceReady,
-					controllertesting.WithChannelServiceReady,
-					controllertesting.WithChannelDeploymentReady,
+					controllertesting.WithReceiverServiceReady,
+					controllertesting.WithReceiverDeploymentReady,
 					controllertesting.WithDispatcherDeploymentReady,
 					controllertesting.WithTopicReady,
 				),
 				controllertesting.NewKafkaChannelService(),
-				controllertesting.NewKafkaChannelChannelService(),
-				controllertesting.NewKafkaChannelChannelDeployment(),
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
 				controllertesting.NewKafkaChannelDispatcherService(),
 			},
 			WantCreates: []runtime.Object{controllertesting.NewKafkaChannelDispatcherDeployment()},
@@ -350,14 +350,14 @@ func TestReconcile(t *testing.T) {
 					controllertesting.WithAddress,
 					controllertesting.WithInitializedConditions,
 					controllertesting.WithKafkaChannelServiceReady,
-					controllertesting.WithChannelServiceReady,
-					controllertesting.WithChannelDeploymentReady,
+					controllertesting.WithReceiverServiceReady,
+					controllertesting.WithReceiverDeploymentReady,
 					controllertesting.WithDispatcherDeploymentReady,
 					controllertesting.WithTopicReady,
 				),
 				controllertesting.NewKafkaChannelService(),
-				controllertesting.NewKafkaChannelChannelService(),
-				controllertesting.NewKafkaChannelChannelDeployment(),
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
 				controllertesting.NewKafkaChannelDispatcherService(),
 			},
 			WithReactors: []clientgotesting.ReactionFunc{InduceFailure("create", "deployments")},
@@ -371,8 +371,8 @@ func TestReconcile(t *testing.T) {
 						controllertesting.WithAddress,
 						controllertesting.WithInitializedConditions,
 						controllertesting.WithKafkaChannelServiceReady,
-						controllertesting.WithChannelServiceReady,
-						controllertesting.WithChannelDeploymentReady,
+						controllertesting.WithReceiverServiceReady,
+						controllertesting.WithReceiverDeploymentReady,
 						controllertesting.WithDispatcherFailed,
 						controllertesting.WithTopicReady,
 					),
diff --git a/pkg/channel/distributed/controller/kafkachannel/topic_test.go b/pkg/channel/distributed/controller/kafkachannel/topic_test.go
index 38ea44517..c984f0caa 100644
--- a/pkg/channel/distributed/controller/kafkachannel/topic_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/topic_test.go
@@ -46,8 +46,8 @@ func TestReconcileTopic(t *testing.T) {
 				controllertesting.WithAddress,
 				controllertesting.WithInitializedConditions,
 				controllertesting.WithKafkaChannelServiceReady,
-				controllertesting.WithChannelServiceReady,
-				controllertesting.WithChannelDeploymentReady,
+				controllertesting.WithReceiverServiceReady,
+				controllertesting.WithReceiverDeploymentReady,
 				controllertesting.WithDispatcherDeploymentReady,
 			),
 			WantCreate: false,
@@ -60,8 +60,8 @@ func TestReconcileTopic(t *testing.T) {
 				controllertesting.WithAddress,
 				controllertesting.WithInitializedConditions,
 				controllertesting.WithKafkaChannelServiceReady,
-				controllertesting.WithChannelServiceReady,
-				controllertesting.WithChannelDeploymentReady,
+				controllertesting.WithReceiverServiceReady,
+				controllertesting.WithReceiverDeploymentReady,
 				controllertesting.WithDispatcherDeploymentReady,
 			),
 			WantCreate: true,
@@ -79,8 +79,8 @@ func TestReconcileTopic(t *testing.T) {
 				controllertesting.WithAddress,
 				controllertesting.WithInitializedConditions,
 				controllertesting.WithKafkaChannelServiceReady,
-				controllertesting.WithChannelServiceReady,
-				controllertesting.WithChannelDeploymentReady,
+				controllertesting.WithReceiverServiceReady,
+				controllertesting.WithReceiverDeploymentReady,
 				controllertesting.WithDispatcherDeploymentReady,
 			),
 			WantCreate: true,
@@ -99,8 +99,8 @@ func TestReconcileTopic(t *testing.T) {
 				controllertesting.WithAddress,
 				controllertesting.WithInitializedConditions,
 				controllertesting.WithKafkaChannelServiceReady,
-				controllertesting.WithChannelServiceReady,
-				controllertesting.WithChannelDeploymentReady,
+				controllertesting.WithReceiverServiceReady,
+				controllertesting.WithReceiverDeploymentReady,
 				controllertesting.WithDispatcherDeploymentReady,
 			),
 			WantCreate: true,
@@ -120,8 +120,8 @@ func TestReconcileTopic(t *testing.T) {
 				controllertesting.WithAddress,
 				controllertesting.WithInitializedConditions,
 				controllertesting.WithKafkaChannelServiceReady,
-				controllertesting.WithChannelServiceReady,
-				controllertesting.WithChannelDeploymentReady,
+				controllertesting.WithReceiverServiceReady,
+				controllertesting.WithReceiverDeploymentReady,
 				controllertesting.WithDispatcherDeploymentReady,
 			),
 			WantCreate: false,
@@ -134,8 +134,8 @@ func TestReconcileTopic(t *testing.T) {
 				controllertesting.WithAddress,
 				controllertesting.WithInitializedConditions,
 				controllertesting.WithKafkaChannelServiceReady,
-				controllertesting.WithChannelServiceReady,
-				controllertesting.WithChannelDeploymentReady,
+				controllertesting.WithReceiverServiceReady,
+				controllertesting.WithReceiverDeploymentReady,
 				controllertesting.WithDispatcherDeploymentReady,
 			),
 			WantCreate:    false,
@@ -149,8 +149,8 @@ func TestReconcileTopic(t *testing.T) {
 				controllertesting.WithAddress,
 				controllertesting.WithInitializedConditions,
 				controllertesting.WithKafkaChannelServiceReady,
-				controllertesting.WithChannelServiceReady,
-				controllertesting.WithChannelDeploymentReady,
+				controllertesting.WithReceiverServiceReady,
+				controllertesting.WithReceiverDeploymentReady,
 				controllertesting.WithDispatcherDeploymentReady,
 			),
 			WantCreate:    false,
@@ -171,7 +171,7 @@ func topicTestCaseFactory(tc TopicTestCase) func(t *testing.T) {
 	return func(t *testing.T) {
 
 		// Setup Context With New Recorder For Testing
-		recorder := record.NewBroadcaster().NewRecorder(scheme.Scheme, corev1.EventSource{Component: constants.KafkaChannelControllerAgentName})
+		recorder := record.NewBroadcaster().NewRecorder(scheme.Scheme, corev1.EventSource{Component: "TestEventSource"})
 		ctx := controller.WithEventRecorder(context.TODO(), recorder)
 
 		// Create A Mock Kafka AdminClient For Current TopicTestCase
diff --git a/pkg/channel/distributed/controller/kafkasecret/controller_test.go b/pkg/channel/distributed/controller/kafkasecret/controller_test.go
index 62bbe12ac..0f8ac61dd 100644
--- a/pkg/channel/distributed/controller/kafkasecret/controller_test.go
+++ b/pkg/channel/distributed/controller/kafkasecret/controller_test.go
@@ -67,6 +67,6 @@ func populateEnvironmentVariables(t *testing.T) {
 	assert.Nil(t, os.Setenv(commonenv.ServiceAccountEnvVarKey, controllertesting.ServiceAccount))
 	assert.Nil(t, os.Setenv(commonenv.MetricsDomainEnvVarKey, controllertesting.MetricsDomain))
 	assert.Nil(t, os.Setenv(commonenv.MetricsPortEnvVarKey, strconv.Itoa(controllertesting.MetricsPort)))
-	assert.Nil(t, os.Setenv(controllerenv.DispatcherImageEnvVarKey, controllertesting.MetricsDomain))
-	assert.Nil(t, os.Setenv(controllerenv.ChannelImageEnvVarKey, controllertesting.MetricsDomain))
+	assert.Nil(t, os.Setenv(controllerenv.DispatcherImageEnvVarKey, controllertesting.DispatcherImage))
+	assert.Nil(t, os.Setenv(controllerenv.ReceiverImageEnvVarKey, controllertesting.ReceiverImage))
 }
diff --git a/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go b/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go
index c4a080da3..704f3a0f4 100644
--- a/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go
+++ b/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go
@@ -66,8 +66,8 @@ func TestReconcile(t *testing.T) {
 				controllertesting.NewKafkaSecret(),
 			},
 			WantCreates: []runtime.Object{
-				controllertesting.NewKafkaChannelChannelService(),
-				controllertesting.NewKafkaChannelChannelDeployment(),
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{controllertesting.NewKafkaSecretFinalizerPatchActionImpl()},
 			WantEvents: []string{
@@ -83,14 +83,14 @@ func TestReconcile(t *testing.T) {
 				controllertesting.NewKafkaChannel(),
 			},
 			WantCreates: []runtime.Object{
-				controllertesting.NewKafkaChannelChannelService(),
-				controllertesting.NewKafkaChannelChannelDeployment(),
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: controllertesting.NewKafkaChannel(
-						controllertesting.WithChannelServiceReady,
-						controllertesting.WithChannelDeploymentReady,
+						controllertesting.WithReceiverServiceReady,
+						controllertesting.WithReceiverDeploymentReady,
 					),
 				},
 			},
@@ -111,15 +111,15 @@ func TestReconcile(t *testing.T) {
 			Objects: []runtime.Object{
 				controllertesting.NewKafkaSecret(controllertesting.WithKafkaSecretDeleted),
 				controllertesting.NewKafkaChannel(
-					controllertesting.WithChannelServiceReady,
-					controllertesting.WithChannelDeploymentReady,
+					controllertesting.WithReceiverServiceReady,
+					controllertesting.WithReceiverDeploymentReady,
 				),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: controllertesting.NewKafkaChannel(
-						controllertesting.WithChannelServiceFinalized,
-						controllertesting.WithChannelDeploymentFinalized,
+						controllertesting.WithReceiverServiceFinalized,
+						controllertesting.WithReceiverDeploymentFinalized,
 					),
 				},
 			},
@@ -129,100 +129,100 @@ func TestReconcile(t *testing.T) {
 		},
 
 		//
-		// KafkaChannel Channel Service
+		// KafkaSecret Receiver Service
 		//
 
 		{
-			Name: "Reconcile Missing Channel Service Success",
+			Name: "Reconcile Missing Receiver Service Success",
 			Key:  controllertesting.KafkaSecretKey,
 			Objects: []runtime.Object{
 				controllertesting.NewKafkaSecret(controllertesting.WithKafkaSecretFinalizer),
 				controllertesting.NewKafkaChannel(
-					controllertesting.WithChannelServiceReady,
-					controllertesting.WithChannelDeploymentReady,
+					controllertesting.WithReceiverServiceReady,
+					controllertesting.WithReceiverDeploymentReady,
 				),
-				controllertesting.NewKafkaChannelChannelDeployment(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
 			},
-			WantCreates: []runtime.Object{controllertesting.NewKafkaChannelChannelService()},
+			WantCreates: []runtime.Object{controllertesting.NewKafkaChannelReceiverService()},
 			WantEvents: []string{
 				controllertesting.NewKafkaSecretSuccessfulReconciliationEvent(),
 			},
 		},
 		{
-			Name: "Reconcile Missing Channel Service Error(Create)",
+			Name: "Reconcile Missing Receiver Service Error(Create)",
 			Key:  controllertesting.KafkaSecretKey,
 			Objects: []runtime.Object{
 				controllertesting.NewKafkaSecret(controllertesting.WithKafkaSecretFinalizer),
 				controllertesting.NewKafkaChannel(
-					controllertesting.WithChannelServiceReady,
-					controllertesting.WithChannelDeploymentReady,
+					controllertesting.WithReceiverServiceReady,
+					controllertesting.WithReceiverDeploymentReady,
 				),
 				controllertesting.NewKafkaChannelService(),
-				controllertesting.NewKafkaChannelChannelDeployment(),
+				controllertesting.NewKafkaChannelReceiverDeployment(),
 				controllertesting.NewKafkaChannelDispatcherService(),
 				controllertesting.NewKafkaChannelDispatcherDeployment(),
 			},
 			WithReactors: []clientgotesting.ReactionFunc{InduceFailure("create", "services")},
 			WantErr:      true,
-			WantCreates:  []runtime.Object{controllertesting.NewKafkaChannelChannelService()},
+			WantCreates:  []runtime.Object{controllertesting.NewKafkaChannelReceiverService()},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: controllertesting.NewKafkaChannel(
-						controllertesting.WithChannelServiceFailed,
-						controllertesting.WithChannelDeploymentReady,
+						controllertesting.WithReceiverServiceFailed,
+						controllertesting.WithReceiverDeploymentReady,
 					),
 				},
 			},
 			WantEvents: []string{
-				Eventf(corev1.EventTypeWarning, event.ChannelServiceReconciliationFailed.String(), "Failed To Reconcile Channel Service: inducing failure for create services"),
+				Eventf(corev1.EventTypeWarning, event.ReceiverServiceReconciliationFailed.String(), "Failed To Reconcile Receiver Service: inducing failure for create services"),
 				controllertesting.NewKafkaSecretFailedReconciliationEvent(),
 			},
 		},
 
 		//
-		// KafkaChannel Channel Deployment
+		// KafkaSecret Receiver Deployment
 		//
 
 		{
-			Name: "Reconcile Missing Channel Deployment Success",
+			Name: "Reconcile Missing Receiver Deployment Success",
 			Key:  controllertesting.KafkaSecretKey,
 			Objects: []runtime.Object{
 				controllertesting.NewKafkaSecret(controllertesting.WithKafkaSecretFinalizer),
 				controllertesting.NewKafkaChannel(
-					controllertesting.WithChannelServiceReady,
-					controllertesting.WithChannelDeploymentReady,
+					controllertesting.WithReceiverServiceReady,
+					controllertesting.WithReceiverDeploymentReady,
 				),
 				controllertesting.NewKafkaChannelService(),
-				controllertesting.NewKafkaChannelChannelService(),
+				controllertesting.NewKafkaChannelReceiverService(),
 			},
-			WantCreates: []runtime.Object{controllertesting.NewKafkaChannelChannelDeployment()},
+			WantCreates: []runtime.Object{controllertesting.NewKafkaChannelReceiverDeployment()},
 			WantEvents:  []string{controllertesting.NewKafkaSecretSuccessfulReconciliationEvent()},
 		},
 		{
-			Name: "Reconcile Missing Channel Deployment Error(Create)",
+			Name: "Reconcile Missing Receiver Deployment Error(Create)",
 			Key:  controllertesting.KafkaSecretKey,
 			Objects: []runtime.Object{
 				controllertesting.NewKafkaSecret(controllertesting.WithKafkaSecretFinalizer),
 				controllertesting.NewKafkaChannel(
-					controllertesting.WithChannelServiceReady,
-					controllertesting.WithChannelDeploymentReady,
+					controllertesting.WithReceiverServiceReady,
+					controllertesting.WithReceiverDeploymentReady,
 				),
 				controllertesting.NewKafkaChannelService(),
-				controllertesting.NewKafkaChannelChannelService(),
+				controllertesting.NewKafkaChannelReceiverService(),
 			},
 			WithReactors: []clientgotesting.ReactionFunc{InduceFailure("create", "deployments")},
 			WantErr:      true,
-			WantCreates:  []runtime.Object{controllertesting.NewKafkaChannelChannelDeployment()},
+			WantCreates:  []runtime.Object{controllertesting.NewKafkaChannelReceiverDeployment()},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: controllertesting.NewKafkaChannel(
-						controllertesting.WithChannelServiceReady,
-						controllertesting.WithChannelDeploymentFailed,
+						controllertesting.WithReceiverServiceReady,
+						controllertesting.WithReceiverDeploymentFailed,
 					),
 				},
 			},
 			WantEvents: []string{
-				Eventf(corev1.EventTypeWarning, event.ChannelDeploymentReconciliationFailed.String(), "Failed To Reconcile Channel Deployment: inducing failure for create deployments"),
+				Eventf(corev1.EventTypeWarning, event.ReceiverDeploymentReconciliationFailed.String(), "Failed To Reconcile Receiver Deployment: inducing failure for create deployments"),
 				controllertesting.NewKafkaSecretFailedReconciliationEvent(),
 			},
 		},
diff --git a/pkg/channel/distributed/controller/testing/data.go b/pkg/channel/distributed/controller/testing/data.go
index 5cca4472f..708ee818d 100644
--- a/pkg/channel/distributed/controller/testing/data.go
+++ b/pkg/channel/distributed/controller/testing/data.go
@@ -36,32 +36,29 @@ const (
 	MetricsPortName = "metrics"
 
 	// Environment Test Data
-	ServiceAccount                         = "TestServiceAccount"
-	KafkaAdminType                         = "kafka"
-	MetricsPort                            = 9876
-	MetricsDomain                          = "eventing-kafka"
-	HealthPort                             = 8082
-	ChannelImage                           = "TestChannelImage"
-	ChannelReplicas                        = 1
-	DispatcherImage                        = "TestDispatcherImage"
-	DispatcherReplicas                     = 1
-	DefaultNumPartitions                   = 4
-	DefaultReplicationFactor               = 1
-	DefaultRetentionMillis                 = 99999
-	DefaultEventRetryInitialIntervalMillis = 88888
-	DefaultEventRetryTimeMillisMax         = 11111111
-	DefaultExponentialBackoff              = true
+	ServiceAccount           = "TestServiceAccount"
+	KafkaAdminType           = "kafka"
+	MetricsPort              = 9876
+	MetricsDomain            = "eventing-kafka"
+	HealthPort               = 8082
+	ReceiverImage            = "TestReceiverImage"
+	ReceiverReplicas         = 1
+	DispatcherImage          = "TestDispatcherImage"
+	DispatcherReplicas       = 1
+	DefaultNumPartitions     = 4
+	DefaultReplicationFactor = 1
+	DefaultRetentionMillis   = 99999
 
 	// Channel Test Data
-	KafkaChannelNamespace = "kafkachannel-namespace"
-	KafkaChannelName      = "kafkachannel-name"
-	KafkaChannelKey       = KafkaChannelNamespace + "/" + KafkaChannelName
-	KafkaSecretNamespace  = commonconstants.KnativeEventingNamespace // Needs To Match Hardcoded Value In Reconciliation
-	KafkaSecretName       = "kafkasecret-name"
-	KafkaSecretKey        = KafkaSecretNamespace + "/" + KafkaSecretName
-	ChannelDeploymentName = KafkaSecretName + "-b9176d5f-channel" // Truncated MD5 Hash Of KafkaSecretName
-	ChannelServiceName    = ChannelDeploymentName
-	TopicName             = KafkaChannelNamespace + "." + KafkaChannelName
+	KafkaChannelNamespace  = "kafkachannel-namespace"
+	KafkaChannelName       = "kafkachannel-name"
+	KafkaChannelKey        = KafkaChannelNamespace + "/" + KafkaChannelName
+	KafkaSecretNamespace   = commonconstants.KnativeEventingNamespace // Needs To Match Hardcoded Value In Reconciliation
+	KafkaSecretName        = "kafkasecret-name"
+	KafkaSecretKey         = KafkaSecretNamespace + "/" + KafkaSecretName
+	ReceiverDeploymentName = KafkaSecretName + "-b9176d5f-receiver" // Truncated MD5 Hash Of KafkaSecretName
+	ReceiverServiceName    = ReceiverDeploymentName
+	TopicName              = KafkaChannelNamespace + "." + KafkaChannelName
 
 	KafkaSecretDataValueBrokers  = "TestKafkaSecretDataBrokers"
 	KafkaSecretDataValueUsername = "TestKafkaSecretDataUsername"
@@ -81,14 +78,14 @@ const (
 	DispatcherMemoryLimit   = "50Mi"
 	DispatcherCpuLimit      = "300m"
 
-	// Test Channel Resources
-	ChannelMemoryRequest = "10Mi"
-	ChannelMemoryLimit   = "20Mi"
-	ChannelCpuRequest    = "10m"
-	ChannelCpuLimit      = "100m"
+	// Test Receiver Resources
+	ReceiverMemoryRequest = "10Mi"
+	ReceiverMemoryLimit   = "20Mi"
+	ReceiverCpuRequest    = "10m"
+	ReceiverCpuLimit      = "100m"
 
 	ControllerConfigYaml = `
-channel:
+receiver:
   cpuLimit: 200m
   cpuRequest: 100m
   memoryLimit: 100Mi
@@ -151,7 +148,7 @@ func NewEnvironment() *env.Environment {
 		MetricsPort:     MetricsPort,
 		MetricsDomain:   MetricsDomain,
 		DispatcherImage: DispatcherImage,
-		ChannelImage:    ChannelImage,
+		ReceiverImage:   ReceiverImage,
 	}
 }
 
@@ -167,13 +164,13 @@ func NewConfig() *config.EventingKafkaConfig {
 				MemoryRequest: resource.MustParse(DispatcherMemoryRequest),
 			},
 		},
-		Channel: config.EKChannelConfig{
+		Receiver: config.EKReceiverConfig{
 			EKKubernetesConfig: config.EKKubernetesConfig{
-				Replicas:      ChannelReplicas,
-				CpuLimit:      resource.MustParse(ChannelCpuLimit),
-				CpuRequest:    resource.MustParse(ChannelCpuRequest),
-				MemoryLimit:   resource.MustParse(ChannelMemoryLimit),
-				MemoryRequest: resource.MustParse(ChannelMemoryRequest),
+				Replicas:      ReceiverReplicas,
+				CpuLimit:      resource.MustParse(ReceiverCpuLimit),
+				CpuRequest:    resource.MustParse(ReceiverCpuRequest),
+				MemoryLimit:   resource.MustParse(ReceiverMemoryLimit),
+				MemoryRequest: resource.MustParse(ReceiverMemoryRequest),
 			},
 		},
 		Kafka: config.EKKafkaConfig{
@@ -363,33 +360,33 @@ func WithKafkaChannelServiceFailed(kafkachannel *kafkav1beta1.KafkaChannel) {
 	kafkachannel.Status.MarkChannelServiceFailed(event.KafkaChannelServiceReconciliationFailed.String(), "Failed To Create KafkaChannel Service: inducing failure for create services")
 }
 
-// Set The KafkaChannel's Channel Service As READY
-func WithChannelServiceReady(kafkachannel *kafkav1beta1.KafkaChannel) {
+// Set The KafkaChannel's Receiver Service As READY
+func WithReceiverServiceReady(kafkachannel *kafkav1beta1.KafkaChannel) {
 	kafkachannel.Status.MarkServiceTrue()
 }
 
-// Set The KafkaChannel's Channel Service As Failed
-func WithChannelServiceFailed(kafkachannel *kafkav1beta1.KafkaChannel) {
-	kafkachannel.Status.MarkServiceFailed(event.ChannelServiceReconciliationFailed.String(), "Channel Service Failed: inducing failure for create services")
+// Set The KafkaChannel's Receiver Service As Failed
+func WithReceiverServiceFailed(kafkachannel *kafkav1beta1.KafkaChannel) {
+	kafkachannel.Status.MarkServiceFailed(event.ReceiverServiceReconciliationFailed.String(), "Receiver Service Failed: inducing failure for create services")
 }
 
-// Set The KafkaChannel's Channel Service As Finalized
-func WithChannelServiceFinalized(kafkachannel *kafkav1beta1.KafkaChannel) {
+// Set The KafkaChannel's Receiver Service As Finalized
+func WithReceiverServiceFinalized(kafkachannel *kafkav1beta1.KafkaChannel) {
 	kafkachannel.Status.MarkServiceFailed("ChannelServiceUnavailable", "Kafka Auth Secret Finalized")
 }
 
-// Set The KafkaChannel's Channel Deployment As READY
-func WithChannelDeploymentReady(kafkachannel *kafkav1beta1.KafkaChannel) {
+// Set The KafkaChannel's Receiver Deployment As READY
+func WithReceiverDeploymentReady(kafkachannel *kafkav1beta1.KafkaChannel) {
 	kafkachannel.Status.MarkEndpointsTrue()
 }
 
-// Set The KafkaChannel's Channel Deployment As Failed
-func WithChannelDeploymentFailed(kafkachannel *kafkav1beta1.KafkaChannel) {
-	kafkachannel.Status.MarkEndpointsFailed(event.ChannelDeploymentReconciliationFailed.String(), "Channel Deployment Failed: inducing failure for create deployments")
+// Set The KafkaChannel's Receiver Deployment As Failed
+func WithReceiverDeploymentFailed(kafkachannel *kafkav1beta1.KafkaChannel) {
+	kafkachannel.Status.MarkEndpointsFailed(event.ReceiverDeploymentReconciliationFailed.String(), "Receiver Deployment Failed: inducing failure for create deployments")
 }
 
-// Set The KafkaChannel's Channel Deployment As Finalized
-func WithChannelDeploymentFinalized(kafkachannel *kafkav1beta1.KafkaChannel) {
+// Set The KafkaChannel's Receiver Deployment As Finalized
+func WithReceiverDeploymentFinalized(kafkachannel *kafkav1beta1.KafkaChannel) {
 	kafkachannel.Status.MarkEndpointsFailed("ChannelDeploymentUnavailable", "Kafka Auth Secret Finalized")
 }
 
@@ -423,7 +420,7 @@ func NewKafkaChannelService() *corev1.Service {
 			Labels: map[string]string{
 				constants.KafkaChannelNameLabel:      KafkaChannelName,
 				constants.KafkaChannelNamespaceLabel: KafkaChannelNamespace,
-				constants.KafkaChannelChannelLabel:   "true",
+				constants.KafkaChannelReceiverLabel:  "true",
 				constants.K8sAppChannelSelectorLabel: constants.K8sAppChannelSelectorValue,
 			},
 			OwnerReferences: []metav1.OwnerReference{
@@ -432,24 +429,24 @@ func NewKafkaChannelService() *corev1.Service {
 		},
 		Spec: corev1.ServiceSpec{
 			Type:         corev1.ServiceTypeExternalName,
-			ExternalName: ChannelDeploymentName + "." + commonconstants.KnativeEventingNamespace + ".svc.cluster.local",
+			ExternalName: ReceiverDeploymentName + "." + commonconstants.KnativeEventingNamespace + ".svc.cluster.local",
 		},
 	}
 }
 
-// Utility Function For Creating A Custom KafkaChannel "Deployment" Service For Testing
-func NewKafkaChannelChannelService() *corev1.Service {
+// Utility Function For Creating A Custom Receiver Service For Testing
+func NewKafkaChannelReceiverService() *corev1.Service {
 	return &corev1.Service{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: corev1.SchemeGroupVersion.String(),
 			Kind:       constants.ServiceKind,
 		},
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      ChannelDeploymentName,
+			Name:      ReceiverDeploymentName,
 			Namespace: commonconstants.KnativeEventingNamespace,
 			Labels: map[string]string{
-				"k8s-app":              "eventing-kafka-channels",
-				"kafkachannel-channel": "true",
+				"k8s-app":               "eventing-kafka-channels",
+				"kafkachannel-receiver": "true",
 			},
 			OwnerReferences: []metav1.OwnerReference{
 				NewSecretOwnerRef(),
@@ -469,26 +466,26 @@ func NewKafkaChannelChannelService() *corev1.Service {
 				},
 			},
 			Selector: map[string]string{
-				"app": ChannelDeploymentName,
+				"app": ReceiverDeploymentName,
 			},
 		},
 	}
 }
 
-// Utility Function For Creating A K8S Channel Deployment For The Test Channel
-func NewKafkaChannelChannelDeployment() *appsv1.Deployment {
-	replicas := int32(ChannelReplicas)
+// Utility Function For Creating A Receiver Deployment For The Test Channel
+func NewKafkaChannelReceiverDeployment() *appsv1.Deployment {
+	replicas := int32(ReceiverReplicas)
 	return &appsv1.Deployment{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: appsv1.SchemeGroupVersion.String(),
 			Kind:       constants.DeploymentKind,
 		},
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      ChannelDeploymentName,
+			Name:      ReceiverDeploymentName,
 			Namespace: commonconstants.KnativeEventingNamespace,
 			Labels: map[string]string{
-				"app":                  ChannelDeploymentName,
-				"kafkachannel-channel": "true",
+				"app":                   ReceiverDeploymentName,
+				"kafkachannel-receiver": "true",
 			},
 			OwnerReferences: []metav1.OwnerReference{
 				NewSecretOwnerRef(),
@@ -498,20 +495,20 @@ func NewKafkaChannelChannelDeployment() *appsv1.Deployment {
 			Replicas: &replicas,
 			Selector: &metav1.LabelSelector{
 				MatchLabels: map[string]string{
-					"app": ChannelDeploymentName,
+					"app": ReceiverDeploymentName,
 				},
 			},
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: map[string]string{
-						"app": ChannelDeploymentName,
+						"app": ReceiverDeploymentName,
 					},
 				},
 				Spec: corev1.PodSpec{
 					ServiceAccountName: ServiceAccount,
 					Containers: []corev1.Container{
 						{
-							Name: ChannelDeploymentName,
+							Name: ReceiverDeploymentName,
 							LivenessProbe: &corev1.Probe{
 								Handler: corev1.Handler{
 									HTTPGet: &corev1.HTTPGetAction{
@@ -532,7 +529,7 @@ func NewKafkaChannelChannelDeployment() *appsv1.Deployment {
 								InitialDelaySeconds: constants.ChannelReadinessDelay,
 								PeriodSeconds:       constants.ChannelReadinessPeriod,
 							},
-							Image: ChannelImage,
+							Image: ReceiverImage,
 							Ports: []corev1.ContainerPort{
 								{
 									Name:          "server",
@@ -550,7 +547,7 @@ func NewKafkaChannelChannelDeployment() *appsv1.Deployment {
 								},
 								{
 									Name:  commonenv.ServiceNameEnvVarKey,
-									Value: ChannelServiceName,
+									Value: ReceiverServiceName,
 								},
 								{
 									Name:  commonenv.MetricsPortEnvVarKey,
@@ -595,12 +592,12 @@ func NewKafkaChannelChannelDeployment() *appsv1.Deployment {
 							ImagePullPolicy: corev1.PullIfNotPresent,
 							Resources: corev1.ResourceRequirements{
 								Requests: corev1.ResourceList{
-									corev1.ResourceCPU:    resource.MustParse(ChannelCpuRequest),
-									corev1.ResourceMemory: resource.MustParse(ChannelMemoryRequest),
+									corev1.ResourceCPU:    resource.MustParse(ReceiverCpuRequest),
+									corev1.ResourceMemory: resource.MustParse(ReceiverMemoryRequest),
 								},
 								Limits: corev1.ResourceList{
-									corev1.ResourceCPU:    resource.MustParse(ChannelCpuLimit),
-									corev1.ResourceMemory: resource.MustParse(ChannelMemoryLimit),
+									corev1.ResourceCPU:    resource.MustParse(ReceiverCpuLimit),
+									corev1.ResourceMemory: resource.MustParse(ReceiverMemoryLimit),
 								},
 							},
 						},
diff --git a/pkg/channel/distributed/controller/util/channel_test.go b/pkg/channel/distributed/controller/util/channel_test.go
index b34d4ee7e..11279895a 100644
--- a/pkg/channel/distributed/controller/util/channel_test.go
+++ b/pkg/channel/distributed/controller/util/channel_test.go
@@ -77,14 +77,14 @@ func TestNewChannelOwnerReference(t *testing.T) {
 	assert.True(t, *controllerRef.Controller)
 }
 
-// Test The ChannelDnsSafeName() Functionality
-func TestChannelDeploymentDnsSafeName(t *testing.T) {
+// Test The ReceiverDnsSafeName() Functionality
+func TestReceiverDnsSafeName(t *testing.T) {
 
 	// Perform The Test
-	actualResult := ChannelDnsSafeName(kafkaSecret)
+	actualResult := ReceiverDnsSafeName(kafkaSecret)
 
 	// Verify The Results
-	expectedResult := fmt.Sprintf("%s-%s-channel", strings.ToLower(kafkaSecret), GenerateHash(kafkaSecret, 8))
+	expectedResult := fmt.Sprintf("%s-%s-receiver", strings.ToLower(kafkaSecret), GenerateHash(kafkaSecret, 8))
 	assert.Equal(t, expectedResult, actualResult)
 }
 
diff --git a/pkg/channel/distributed/receiver/testing/constants.go b/pkg/channel/distributed/receiver/testing/constants.go
index cf4d3acf6..fbbf78b11 100644
--- a/pkg/channel/distributed/receiver/testing/constants.go
+++ b/pkg/channel/distributed/receiver/testing/constants.go
@@ -11,7 +11,7 @@ const (
 	ChannelName      = "TestChannelName"
 	ChannelNamespace = "TestChannelNamespace"
 
-	TopicName = "TestChannelNamespace.TestChannelName" // Match util.go TopicName() Implementation For Channel Name/Namespace Abovce
+	TopicName = "TestChannelNamespace.TestChannelName" // Match util.go TopicName() Implementation For Channel Name/Namespace Above
 
 	PartitionKey         = "TestPartitionKey"
 	EventId              = "TestEventId"