Repository: knative-sandbox/eventing-kafka
Pull request #275 (2020-12-11 18:39:26)

Remove default limits
## Proposed Changes

- 🎁 Allow not having the cpu/memory resource requests and/or limits in the config-kafka configmap for eventing-kafka.receiver and eventing-kafka.dispatcher
- 🧽 Default to no limit for receiver and dispatcher (but still allow a limit if defined by the admin in the config-kafka configmap) and a reasonable (100m / 50Mi) initial request size based on empirical small-cluster and small-message-size testing.


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/275#" title="Author self-approved">eric-sap</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **phamilton** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @phamilton` in a comment when ready.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["phamilton"]} -->
knative-prow-robot: Hi @eric-sap. Thanks for your PR.

I'm waiting for a [knative-sandbox](https://github.com/orgs/knative-sandbox/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/knative-sandbox/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/275?src=pr&el=h1) Report
> Merging [#275](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/275?src=pr&el=desc) (73d8160) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/8f02afa7b1d5cc5838dd69ae10d4cf0811e6b4a4?el=desc) (8f02afa) will **decrease** coverage by `0.06%`.
> The diff coverage is `100.00%`.

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

```diff
@@            Coverage Diff             @@
##           master     #275      +/-   ##
==========================================
- Coverage   75.67%   75.61%   -0.07%     
==========================================
  Files         114      114              
  Lines        4481     4478       -3     
==========================================
- Hits         3391     3386       -5     
- Misses        888      889       +1     
- Partials      202      203       +1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/275?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...kg/channel/distributed/controller/config/config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/275/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9jb25maWcvY29uZmlnLmdv) | `96.29% <100.00%> (-0.85%)` | :arrow_down: |
| [.../distributed/controller/kafkachannel/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/275/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYWNoYW5uZWwvZGlzcGF0Y2hlci5nbw==) | `90.15% <100.00%> (+0.06%)` | :arrow_up: |
| [...nel/distributed/controller/kafkasecret/receiver.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/275/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYXNlY3JldC9yZWNlaXZlci5nbw==) | `93.38% <100.00%> (+0.05%)` | :arrow_up: |
| [pkg/source/adapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/275/diff?src=pr&el=tree#diff-cGtnL3NvdXJjZS9hZGFwdGVyL2FkYXB0ZXIuZ28=) | `70.90% <100.00%> (+0.53%)` | :arrow_up: |
| [...tributed/common/kafka/admin/eventhubcache/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/275/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29tbW9uL2thZmthL2FkbWluL2V2ZW50aHViY2FjaGUvY2FjaGUuZ28=) | `65.06% <0.00%> (-2.41%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/275?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/275?src=pr&el=footer). Last update [8f02afa...617faa8](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/275?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

matzew: /ok-to-test
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/config/config.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338357630711631872/artifacts/line-cov.html#file1) | 95.0% | 93.8% | -1.2
[pkg/channel/distributed/controller/kafkachannel/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338357630711631872/artifacts/line-cov.html#file2) | 86.4% | 86.9% | 0.5
[pkg/channel/distributed/controller/kafkasecret/receiver.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338357630711631872/artifacts/line-cov.html#file3) | 87.3% | 88.1% | 0.8

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/config/config.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338544380666974208/artifacts/line-cov.html#file1) | 95.0% | 91.7% | -3.3
[pkg/channel/distributed/controller/kafkachannel/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338544380666974208/artifacts/line-cov.html#file2) | 86.4% | 87.3% | 0.9
[pkg/channel/distributed/controller/kafkasecret/receiver.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338544380666974208/artifacts/line-cov.html#file4) | 87.3% | 88.8% | 1.4

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/config/config.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338546775635529728/artifacts/line-cov.html#file1) | 95.0% | 91.7% | -3.3
[pkg/channel/distributed/controller/kafkachannel/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338546775635529728/artifacts/line-cov.html#file2) | 86.4% | 87.3% | 0.9
[pkg/channel/distributed/controller/kafkasecret/receiver.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338546775635529728/artifacts/line-cov.html#file4) | 87.3% | 88.8% | 1.4

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/config/config.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338552597518225408/artifacts/line-cov.html#file1) | 95.0% | 91.7% | -3.3
[pkg/channel/distributed/controller/kafkachannel/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338552597518225408/artifacts/line-cov.html#file2) | 86.4% | 87.3% | 0.9
[pkg/channel/distributed/controller/kafkasecret/receiver.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338552597518225408/artifacts/line-cov.html#file4) | 87.3% | 88.8% | 1.4

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/config/config.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338561503749476352/artifacts/line-cov.html#file1) | 95.0% | 91.7% | -3.3
[pkg/channel/distributed/controller/kafkachannel/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338561503749476352/artifacts/line-cov.html#file2) | 86.4% | 87.3% | 0.9
[pkg/channel/distributed/controller/kafkasecret/receiver.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/275/pull-knative-sandbox-eventing-kafka-go-coverage/1338561503749476352/artifacts/line-cov.html#file4) | 87.3% | 88.8% | 1.4

travis-minke-sap: /approve
/lgtm
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/275#" title="Author self-approved">eric-sap</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/275#issuecomment-744666728" title="LGTM">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":[]} -->

Patch:
diff --git a/config/channel/distributed/300-eventing-kafka-configmap.yaml b/config/channel/distributed/300-eventing-kafka-configmap.yaml
index f46fd7d7b..59a726455 100644
--- a/config/channel/distributed/300-eventing-kafka-configmap.yaml
+++ b/config/channel/distributed/300-eventing-kafka-configmap.yaml
@@ -39,15 +39,11 @@ data:
       RequiredAcks: -1  # -1 = WaitForAll, Most stringent option for "at-least-once" delivery.
   eventing-kafka: |
     receiver:
-      cpuLimit: 200m
       cpuRequest: 100m
-      memoryLimit: 100Mi
       memoryRequest: 50Mi
       replicas: 1
     dispatcher:
-      cpuLimit: 500m
-      cpuRequest: 300m
-      memoryLimit: 128Mi
+      cpuRequest: 100m
       memoryRequest: 50Mi
       replicas: 1
     kafka:
diff --git a/config/channel/distributed/500-controller-deployment.yaml b/config/channel/distributed/500-controller-deployment.yaml
index e30cdef13..1ba7d8b30 100644
--- a/config/channel/distributed/500-controller-deployment.yaml
+++ b/config/channel/distributed/500-controller-deployment.yaml
@@ -69,6 +69,3 @@ spec:
           requests:
             cpu: 20m
             memory: 25Mi
-          limits:
-            cpu: 100m
-            memory: 50Mi
diff --git a/pkg/channel/distributed/controller/config/config.go b/pkg/channel/distributed/controller/config/config.go
index 7742ff408..a438b9c0d 100644
--- a/pkg/channel/distributed/controller/config/config.go
+++ b/pkg/channel/distributed/controller/config/config.go
@@ -19,7 +19,6 @@ package config
 import (
 	"strings"
 
-	"k8s.io/apimachinery/pkg/api/resource"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
 )
@@ -53,24 +52,8 @@ func VerifyConfiguration(configuration *config.EventingKafkaConfig) error {
 		return ControllerConfigurationError("Kafka.Topic.DefaultReplicationFactor must be > 0")
 	case configuration.Kafka.Topic.DefaultRetentionMillis < 1:
 		return ControllerConfigurationError("Kafka.Topic.DefaultRetentionMillis must be > 0")
-	case configuration.Dispatcher.CpuLimit == resource.Quantity{}:
-		return ControllerConfigurationError("Dispatcher.CpuLimit must be nonzero")
-	case configuration.Dispatcher.CpuRequest == resource.Quantity{}:
-		return ControllerConfigurationError("Dispatcher.CpuRequest must be nonzero")
-	case configuration.Dispatcher.MemoryLimit == resource.Quantity{}:
-		return ControllerConfigurationError("Dispatcher.MemoryLimit must be nonzero")
-	case configuration.Dispatcher.MemoryRequest == resource.Quantity{}:
-		return ControllerConfigurationError("Dispatcher.MemoryRequest must be nonzero")
 	case configuration.Dispatcher.Replicas < 1:
 		return ControllerConfigurationError("Dispatcher.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")
 	}
diff --git a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
index 5596efb78..548289c16 100644
--- a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
+++ b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
@@ -22,6 +22,8 @@ import (
 	"strconv"
 	"time"
 
+	"k8s.io/apimachinery/pkg/api/resource"
+
 	"go.uber.org/zap"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -382,6 +384,25 @@ func (r *Reconciler) newDispatcherDeployment(logger *zap.Logger, channel *kafkav
 		return nil, err
 	}
 
+	// There is a difference between setting an entry in the limits or requests map to the zero-value
+	// of a Quantity and not actually having that entry in the map at all.
+	// If we want "no limit" or "no request" then the entry must not be present in the map.
+	// Note: Since a "Quantity" type has no nil value, we use the Zero value to represent unlimited.
+	resourceLimits := make(map[corev1.ResourceName]resource.Quantity)
+	if !r.config.Dispatcher.MemoryLimit.IsZero() {
+		resourceLimits[corev1.ResourceMemory] = r.config.Dispatcher.MemoryLimit
+	}
+	if !r.config.Dispatcher.CpuLimit.IsZero() {
+		resourceLimits[corev1.ResourceCPU] = r.config.Dispatcher.CpuLimit
+	}
+	resourceRequests := make(map[corev1.ResourceName]resource.Quantity)
+	if !r.config.Dispatcher.MemoryRequest.IsZero() {
+		resourceRequests[corev1.ResourceMemory] = r.config.Dispatcher.MemoryRequest
+	}
+	if !r.config.Dispatcher.CpuRequest.IsZero() {
+		resourceRequests[corev1.ResourceCPU] = r.config.Dispatcher.CpuRequest
+	}
+
 	// Create The Dispatcher's Deployment
 	deployment := &appsv1.Deployment{
 		TypeMeta: metav1.TypeMeta{
@@ -443,14 +464,8 @@ func (r *Reconciler) newDispatcherDeployment(logger *zap.Logger, channel *kafkav
 							Env:             envVars,
 							ImagePullPolicy: corev1.PullIfNotPresent,
 							Resources: corev1.ResourceRequirements{
-								Limits: corev1.ResourceList{
-									corev1.ResourceMemory: r.config.Dispatcher.MemoryLimit,
-									corev1.ResourceCPU:    r.config.Dispatcher.CpuLimit,
-								},
-								Requests: corev1.ResourceList{
-									corev1.ResourceMemory: r.config.Dispatcher.MemoryRequest,
-									corev1.ResourceCPU:    r.config.Dispatcher.CpuRequest,
-								},
+								Limits:   resourceLimits,
+								Requests: resourceRequests,
 							},
 						},
 					},
diff --git a/pkg/channel/distributed/controller/kafkasecret/receiver.go b/pkg/channel/distributed/controller/kafkasecret/receiver.go
index 50c868a10..726716d4b 100644
--- a/pkg/channel/distributed/controller/kafkasecret/receiver.go
+++ b/pkg/channel/distributed/controller/kafkasecret/receiver.go
@@ -22,6 +22,8 @@ import (
 	"strconv"
 	"time"
 
+	"k8s.io/apimachinery/pkg/api/resource"
+
 	"go.uber.org/zap"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -262,6 +264,25 @@ func (r *Reconciler) newReceiverDeployment(logger *zap.Logger, secret *corev1.Se
 		return nil, err
 	}
 
+	// There is a difference between setting an entry in the limits or requests map to the zero-value
+	// of a Quantity and not actually having that entry in the map at all.
+	// If we want "no limit" or "no request" then the entry must not be present in the map.
+	// Note: Since a "Quantity" type has no nil value, we use the Zero value to represent unlimited.
+	resourceLimits := make(map[corev1.ResourceName]resource.Quantity)
+	if !r.config.Receiver.MemoryLimit.IsZero() {
+		resourceLimits[corev1.ResourceMemory] = r.config.Receiver.MemoryLimit
+	}
+	if !r.config.Receiver.CpuLimit.IsZero() {
+		resourceLimits[corev1.ResourceCPU] = r.config.Receiver.CpuLimit
+	}
+	resourceRequests := make(map[corev1.ResourceName]resource.Quantity)
+	if !r.config.Receiver.MemoryRequest.IsZero() {
+		resourceRequests[corev1.ResourceMemory] = r.config.Receiver.MemoryRequest
+	}
+	if !r.config.Receiver.CpuRequest.IsZero() {
+		resourceRequests[corev1.ResourceCPU] = r.config.Receiver.CpuRequest
+	}
+
 	// Create The Receiver Deployment
 	deployment := &appsv1.Deployment{
 		TypeMeta: metav1.TypeMeta{
@@ -327,14 +348,8 @@ func (r *Reconciler) newReceiverDeployment(logger *zap.Logger, secret *corev1.Se
 							Env:             channelEnvVars,
 							ImagePullPolicy: corev1.PullIfNotPresent,
 							Resources: corev1.ResourceRequirements{
-								Requests: corev1.ResourceList{
-									corev1.ResourceCPU:    r.config.Receiver.CpuRequest,
-									corev1.ResourceMemory: r.config.Receiver.MemoryRequest,
-								},
-								Limits: corev1.ResourceList{
-									corev1.ResourceCPU:    r.config.Receiver.CpuLimit,
-									corev1.ResourceMemory: r.config.Receiver.MemoryLimit,
-								},
+								Requests: resourceRequests,
+								Limits:   resourceLimits,
 							},
 						},
 					},


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 a2710bb40..6995c986c 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
@@ -39,15 +39,11 @@ const (
 	// EKDefaultConfigYaml is intended to match what's in 200-eventing-kafka-configmap.yaml
 	EKDefaultConfigYaml = `
 receiver:
-  cpuLimit: 200m
   cpuRequest: 100m
-  memoryLimit: 100Mi
   memoryRequest: 50Mi
   replicas: 1
 dispatcher:
-  cpuLimit: 500m
-  cpuRequest: 300m
-  memoryLimit: 128Mi
+  cpuRequest: 100m
   memoryRequest: 50Mi
   replicas: 1
 kafka:
@@ -214,17 +210,17 @@ 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.Receiver.CpuLimit)
+	assert.Equal(t, resource.Quantity{}, 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.Quantity{}, 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)
-	assert.Equal(t, resource.MustParse("500m"), configuration.Dispatcher.CpuLimit)
-	assert.Equal(t, resource.MustParse("300m"), configuration.Dispatcher.CpuRequest)
-	assert.Equal(t, resource.MustParse("128Mi"), configuration.Dispatcher.MemoryLimit)
+	assert.Equal(t, resource.Quantity{}, configuration.Dispatcher.CpuLimit)
+	assert.Equal(t, resource.MustParse("100m"), configuration.Dispatcher.CpuRequest)
+	assert.Equal(t, resource.Quantity{}, configuration.Dispatcher.MemoryLimit)
 	assert.Equal(t, resource.MustParse("50Mi"), configuration.Dispatcher.MemoryRequest)
 	assert.Equal(t, 1, configuration.Dispatcher.Replicas)
 	assert.Equal(t, "azure", configuration.Kafka.AdminType)
diff --git a/pkg/channel/distributed/controller/config/config_test.go b/pkg/channel/distributed/controller/config/config_test.go
index 9f408d470..3f35be7e2 100644
--- a/pkg/channel/distributed/controller/config/config_test.go
+++ b/pkg/channel/distributed/controller/config/config_test.go
@@ -38,11 +38,11 @@ const (
 	dispatcherMemoryLimit   = "50Mi"
 	dispatcherCpuLimit      = "300m"
 
-	channelReplicas      = 1
-	channelMemoryRequest = "10Mi"
-	channelCpuRquest     = "10m"
-	channelMemoryLimit   = "20Mi"
-	channelCpuLimit      = "100m"
+	receiverReplicas      = 1
+	receiverMemoryRequest = "10Mi"
+	receiverCpuRquest     = "10m"
+	receiverMemoryLimit   = "20Mi"
+	receiverCpuLimit      = "100m"
 )
 
 // Define The TestCase Struct
@@ -59,11 +59,11 @@ type TestCase struct {
 	dispatcherMemoryLimit              resource.Quantity
 	dispatcherMemoryRequest            resource.Quantity
 	dispatcherReplicas                 int
-	channelCpuLimit                    resource.Quantity
-	channelCpuRequest                  resource.Quantity
-	channelMemoryLimit                 resource.Quantity
-	channelMemoryRequest               resource.Quantity
-	channelReplicas                    int
+	receiverCpuLimit                   resource.Quantity
+	receiverCpuRequest                 resource.Quantity
+	receiverMemoryLimit                resource.Quantity
+	receiverMemoryRequest              resource.Quantity
+	receiverReplicas                   int
 
 	expectedError error
 }
@@ -81,11 +81,11 @@ func getValidTestCase(name string) TestCase {
 		dispatcherMemoryLimit:              resource.MustParse(dispatcherMemoryLimit),
 		dispatcherMemoryRequest:            resource.MustParse(dispatcherMemoryRequest),
 		dispatcherReplicas:                 dispatcherReplicas,
-		channelCpuLimit:                    resource.MustParse(channelCpuLimit),
-		channelCpuRequest:                  resource.MustParse(channelCpuRquest),
-		channelMemoryLimit:                 resource.MustParse(channelMemoryLimit),
-		channelMemoryRequest:               resource.MustParse(channelMemoryRequest),
-		channelReplicas:                    channelReplicas,
+		receiverCpuLimit:                   resource.MustParse(receiverCpuLimit),
+		receiverCpuRequest:                 resource.MustParse(receiverCpuRquest),
+		receiverMemoryLimit:                resource.MustParse(receiverMemoryLimit),
+		receiverMemoryRequest:              resource.MustParse(receiverMemoryRequest),
+		receiverReplicas:                   receiverReplicas,
 		expectedError:                      nil,
 	}
 }
@@ -114,24 +114,20 @@ func TestVerifyConfiguration(t *testing.T) {
 	testCase.expectedError = ControllerConfigurationError("Kafka.Topic.DefaultRetentionMillis must be > 0")
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Dispatcher.CpuLimit")
+	testCase = getValidTestCase("Valid Config - Dispatcher.CpuLimit = Zero (unlimited)")
 	testCase.dispatcherCpuLimit = resource.Quantity{}
-	testCase.expectedError = ControllerConfigurationError("Dispatcher.CpuLimit must be nonzero")
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Dispatcher.CpuRequest")
-	testCase.dispatcherCpuRequest = resource.Quantity{}
-	testCase.expectedError = ControllerConfigurationError("Dispatcher.CpuRequest must be nonzero")
+	testCase = getValidTestCase("Valid Config - Dispatcher.MemoryLimit = Zero (unlimited)")
+	testCase.dispatcherMemoryLimit = resource.Quantity{}
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Dispatcher.MemoryLimit")
-	testCase.dispatcherMemoryLimit = resource.Quantity{}
-	testCase.expectedError = ControllerConfigurationError("Dispatcher.MemoryLimit must be nonzero")
+	testCase = getValidTestCase("Valid Config - Dispatcher.CpuRequest = Zero (unlimited)")
+	testCase.dispatcherCpuRequest = resource.Quantity{}
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Dispatcher.MemoryRequest")
+	testCase = getValidTestCase("Valid Config - Dispatcher.MemoryRequest = Zero (unlimited)")
 	testCase.dispatcherMemoryRequest = resource.Quantity{}
-	testCase.expectedError = ControllerConfigurationError("Dispatcher.MemoryRequest must be nonzero")
 	testCases = append(testCases, testCase)
 
 	testCase = getValidTestCase("Invalid Config - Dispatcher.Replicas")
@@ -139,28 +135,24 @@ func TestVerifyConfiguration(t *testing.T) {
 	testCase.expectedError = ControllerConfigurationError("Dispatcher.Replicas must be > 0")
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Receiver.CpuLimit")
-	testCase.channelCpuLimit = resource.Quantity{}
-	testCase.expectedError = ControllerConfigurationError("Receiver.CpuLimit must be nonzero")
+	testCase = getValidTestCase("Valid Config - Receiver.CpuLimit = Zero (unlimited)")
+	testCase.receiverCpuLimit = resource.Quantity{}
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Receiver.CpuRequest")
-	testCase.channelCpuRequest = resource.Quantity{}
-	testCase.expectedError = ControllerConfigurationError("Receiver.CpuRequest must be nonzero")
+	testCase = getValidTestCase("Valid Config - Receiver.MemoryLimit = Zero (unlimited)")
+	testCase.receiverMemoryLimit = resource.Quantity{}
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Receiver.MemoryLimit")
-	testCase.channelMemoryLimit = resource.Quantity{}
-	testCase.expectedError = ControllerConfigurationError("Receiver.MemoryLimit must be nonzero")
+	testCase = getValidTestCase("Valid Config - Receiver.CpuRequest = Zero (unlimited)")
+	testCase.receiverCpuRequest = resource.Quantity{}
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Receiver.MemoryRequest")
-	testCase.channelMemoryRequest = resource.Quantity{}
-	testCase.expectedError = ControllerConfigurationError("Receiver.MemoryRequest must be nonzero")
+	testCase = getValidTestCase("Valid Config - Receiver.MemoryRequest = Zero (unlimited)")
+	testCase.receiverMemoryRequest = resource.Quantity{}
 	testCases = append(testCases, testCase)
 
 	testCase = getValidTestCase("Invalid Config - Receiver.Replicas")
-	testCase.channelReplicas = -1
+	testCase.receiverReplicas = -1
 	testCase.expectedError = ControllerConfigurationError("Receiver.Replicas must be > 0")
 	testCases = append(testCases, testCase)
 
@@ -181,11 +173,11 @@ func TestVerifyConfiguration(t *testing.T) {
 			testConfig.Dispatcher.MemoryLimit = testCase.dispatcherMemoryLimit
 			testConfig.Dispatcher.MemoryRequest = testCase.dispatcherMemoryRequest
 			testConfig.Dispatcher.Replicas = testCase.dispatcherReplicas
-			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
+			testConfig.Receiver.CpuLimit = testCase.receiverCpuLimit
+			testConfig.Receiver.CpuRequest = testCase.receiverCpuRequest
+			testConfig.Receiver.MemoryLimit = testCase.receiverMemoryLimit
+			testConfig.Receiver.MemoryRequest = testCase.receiverMemoryRequest
+			testConfig.Receiver.Replicas = testCase.receiverReplicas
 
 			// Perform The Test
 			err := VerifyConfiguration(testConfig)
@@ -202,11 +194,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.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)
+				assert.Equal(t, testCase.receiverCpuLimit, testConfig.Receiver.CpuLimit)
+				assert.Equal(t, testCase.receiverCpuRequest, testConfig.Receiver.CpuRequest)
+				assert.Equal(t, testCase.receiverMemoryLimit, testConfig.Receiver.MemoryLimit)
+				assert.Equal(t, testCase.receiverMemoryRequest, testConfig.Receiver.MemoryRequest)
+				assert.Equal(t, testCase.receiverReplicas, testConfig.Receiver.Replicas)
 			} else {
 				assert.Equal(t, testCase.expectedError, err)
 			}
diff --git a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
index 41ff9bd5a..09060ea52 100644
--- a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
@@ -190,6 +190,51 @@ func TestReconcile(t *testing.T) {
 				controllertesting.NewKafkaChannelSuccessfulReconciliationEvent(),
 			},
 		},
+		{
+			Name:                    "Complete Reconciliation Success, No Dispatcher Resource Requests Or Limits",
+			SkipNamespaceValidation: true,
+			Key:                     controllertesting.KafkaChannelKey,
+			Objects: []runtime.Object{
+				controllertesting.NewKafkaChannel(controllertesting.WithInitializedConditions),
+			},
+			WantCreates: []runtime.Object{
+				controllertesting.NewKafkaChannelService(),
+				controllertesting.NewKafkaChannelDispatcherService(),
+				controllertesting.NewKafkaChannelDispatcherDeployment(controllertesting.WithoutResources),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: controllertesting.NewKafkaChannel(
+						controllertesting.WithAddress,
+						controllertesting.WithInitializedConditions,
+						controllertesting.WithKafkaChannelServiceReady,
+						controllertesting.WithDispatcherDeploymentReady,
+						controllertesting.WithTopicReady,
+					),
+				},
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				controllertesting.NewKafkaChannelLabelUpdate(
+					controllertesting.NewKafkaChannel(
+						controllertesting.WithFinalizer,
+						controllertesting.WithMetaData,
+						controllertesting.WithAddress,
+						controllertesting.WithInitializedConditions,
+						controllertesting.WithKafkaChannelServiceReady,
+						controllertesting.WithDispatcherDeploymentReady,
+						controllertesting.WithTopicReady,
+					),
+				),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{controllertesting.NewFinalizerPatchActionImpl()},
+			WantEvents: []string{
+				controllertesting.NewKafkaChannelFinalizerUpdateEvent(),
+				controllertesting.NewKafkaChannelSuccessfulReconciliationEvent(),
+			},
+			OtherTestData: map[string]interface{}{
+				"configOptions": []controllertesting.KafkaConfigOption{controllertesting.WithNoDispatcherResources},
+			},
+		},
 
 		//
 		// KafkaChannel Deletion (Finalizer)
@@ -599,14 +644,14 @@ func TestReconcile(t *testing.T) {
 
 	// Run The TableTest Using The KafkaChannel Reconciler Provided By The Factory
 	logger := logtesting.TestLogger(t)
-	tableTest.Test(t, controllertesting.MakeFactory(func(ctx context.Context, listers *controllertesting.Listers, cmw configmap.Watcher) controller.Reconciler {
+	tableTest.Test(t, controllertesting.MakeFactory(func(ctx context.Context, listers *controllertesting.Listers, cmw configmap.Watcher, configOptions []controllertesting.KafkaConfigOption) controller.Reconciler {
 		r := &Reconciler{
 			logger:               logging.FromContext(ctx).Desugar(),
 			kubeClientset:        kubeclient.Get(ctx),
 			adminClientType:      kafkaadmin.Kafka,
 			adminClient:          nil,
 			environment:          controllertesting.NewEnvironment(),
-			config:               controllertesting.NewConfig(),
+			config:               controllertesting.NewConfig(configOptions...),
 			kafkachannelLister:   listers.GetKafkaChannelLister(),
 			kafkachannelInformer: nil,
 			deploymentLister:     listers.GetDeploymentLister(),
diff --git a/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go b/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go
index cf9e67ae2..be0ee1528 100644
--- a/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go
+++ b/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go
@@ -119,6 +119,34 @@ func TestReconcile(t *testing.T) {
 				controllertesting.NewKafkaSecretSuccessfulReconciliationEvent(),
 			},
 		},
+		{
+			Name: "Complete Reconciliation With KafkaChannel, No Receiver Resource Requests Or Limits",
+			Key:  controllertesting.KafkaSecretKey,
+			Objects: []runtime.Object{
+				controllertesting.NewKafkaSecret(),
+				controllertesting.NewKafkaChannel(),
+			},
+			WantCreates: []runtime.Object{
+				controllertesting.NewKafkaChannelReceiverService(),
+				controllertesting.NewKafkaChannelReceiverDeployment(controllertesting.WithoutResources),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: controllertesting.NewKafkaChannel(
+						controllertesting.WithReceiverServiceReady,
+						controllertesting.WithReceiverDeploymentReady,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{controllertesting.NewKafkaSecretFinalizerPatchActionImpl()},
+			WantEvents: []string{
+				controllertesting.NewKafkaSecretFinalizerUpdateEvent(),
+				controllertesting.NewKafkaSecretSuccessfulReconciliationEvent(),
+			},
+			OtherTestData: map[string]interface{}{
+				"configOptions": []controllertesting.KafkaConfigOption{controllertesting.WithNoReceiverResources},
+			},
+		},
 
 		//
 		// KafkaChannel Secret Deletion (Finalizer)
@@ -277,12 +305,12 @@ func TestReconcile(t *testing.T) {
 
 	// Run The TableTest Using The KafkaChannel Reconciler Provided By The Factory
 	logger := logtesting.TestLogger(t)
-	tableTest.Test(t, controllertesting.MakeFactory(func(ctx context.Context, listers *controllertesting.Listers, cmw configmap.Watcher) controller.Reconciler {
+	tableTest.Test(t, controllertesting.MakeFactory(func(ctx context.Context, listers *controllertesting.Listers, cmw configmap.Watcher, configOptions []controllertesting.KafkaConfigOption) controller.Reconciler {
 		r := &Reconciler{
 			logger:             logging.FromContext(ctx).Desugar(),
 			kubeClientset:      kubeclient.Get(ctx),
 			environment:        controllertesting.NewEnvironment(),
-			config:             controllertesting.NewConfig(),
+			config:             controllertesting.NewConfig(configOptions...),
 			kafkaChannelClient: fakekafkaclient.Get(ctx),
 			kafkachannelLister: listers.GetKafkaChannelLister(),
 			deploymentLister:   listers.GetDeploymentLister(),
diff --git a/pkg/channel/distributed/controller/testing/data.go b/pkg/channel/distributed/controller/testing/data.go
index 998e4dd47..b7de7ab8f 100644
--- a/pkg/channel/distributed/controller/testing/data.go
+++ b/pkg/channel/distributed/controller/testing/data.go
@@ -185,6 +185,11 @@ func WithoutFinalizersDeployment(deployment *appsv1.Deployment) {
 	deployment.ObjectMeta.Finalizers = []string{}
 }
 
+func WithoutResources(deployment *appsv1.Deployment) {
+	deployment.Spec.Template.Spec.Containers[0].Resources.Limits = nil
+	deployment.Spec.Template.Spec.Containers[0].Resources.Requests = nil
+}
+
 //
 // ControllerConfig Test Data
 //
@@ -202,9 +207,12 @@ func NewEnvironment() *env.Environment {
 	}
 }
 
+// KafkaConfigOption Enables Customization Of An EventingKafkaConfig
+type KafkaConfigOption func(kafkaConfig *config.EventingKafkaConfig)
+
 // Set The Required Config Fields
-func NewConfig() *config.EventingKafkaConfig {
-	return &config.EventingKafkaConfig{
+func NewConfig(options ...KafkaConfigOption) *config.EventingKafkaConfig {
+	kafkaConfig := &config.EventingKafkaConfig{
 		Dispatcher: config.EKDispatcherConfig{
 			EKKubernetesConfig: config.EKKubernetesConfig{
 				Replicas:      DispatcherReplicas,
@@ -232,6 +240,29 @@ func NewConfig() *config.EventingKafkaConfig {
 			AdminType: KafkaAdminType,
 		},
 	}
+
+	// Apply The Specified Kafka secret Customizations
+	for _, option := range options {
+		option(kafkaConfig)
+	}
+
+	return kafkaConfig
+}
+
+// Remove The Receiver Resource Requests And Limits
+func WithNoReceiverResources(kafkaConfig *config.EventingKafkaConfig) {
+	kafkaConfig.Receiver.EKKubernetesConfig.CpuLimit = resource.Quantity{}
+	kafkaConfig.Receiver.EKKubernetesConfig.CpuRequest = resource.Quantity{}
+	kafkaConfig.Receiver.EKKubernetesConfig.MemoryLimit = resource.Quantity{}
+	kafkaConfig.Receiver.EKKubernetesConfig.MemoryRequest = resource.Quantity{}
+}
+
+// Remove The Dispatcher Resource Requests And Limits
+func WithNoDispatcherResources(kafkaConfig *config.EventingKafkaConfig) {
+	kafkaConfig.Dispatcher.EKKubernetesConfig.CpuLimit = resource.Quantity{}
+	kafkaConfig.Dispatcher.EKKubernetesConfig.CpuRequest = resource.Quantity{}
+	kafkaConfig.Dispatcher.EKKubernetesConfig.MemoryLimit = resource.Quantity{}
+	kafkaConfig.Dispatcher.EKKubernetesConfig.MemoryRequest = resource.Quantity{}
 }
 
 //
diff --git a/pkg/channel/distributed/controller/testing/factory.go b/pkg/channel/distributed/controller/testing/factory.go
index 14a95b4b4..3433e8528 100644
--- a/pkg/channel/distributed/controller/testing/factory.go
+++ b/pkg/channel/distributed/controller/testing/factory.go
@@ -43,7 +43,7 @@ const (
 )
 
 // Ctor functions create a k8s controller with given params.
-type Ctor func(context.Context, *Listers, configmap.Watcher) controller.Reconciler
+type Ctor func(context.Context, *Listers, configmap.Watcher, []KafkaConfigOption) controller.Reconciler
 
 // MakeFactory creates a reconciler factory with fake clients and controller created by `ctor`.
 func MakeFactory(ctor Ctor, logger *zap.Logger) Factory {
@@ -67,8 +67,13 @@ func MakeFactory(ctor Ctor, logger *zap.Logger) Factory {
 		eventRecorder := record.NewFakeRecorder(maxEventBufferSize)
 		ctx = controller.WithEventRecorder(ctx, eventRecorder)
 
+		configOptions, ok := r.OtherTestData["configOptions"]
+		if !ok || configOptions == nil {
+			configOptions = []KafkaConfigOption{}
+		}
+
 		// Set up our Controller from the fakes.
-		c := ctor(ctx, &ls, configmap.NewStaticWatcher())
+		c := ctor(ctx, &ls, configmap.NewStaticWatcher(), configOptions.([]KafkaConfigOption))
 
 		// The Reconciler won't do any work until it becomes the leader.
 		if la, ok := c.(reconciler.LeaderAware); ok {