Repository: knative-sandbox/eventing-kafka
Pull request #470 (2021-03-16 20:02:07)

[mtsource] Automatically adjust Sarama resource usage to fit within limits
<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Compute the maximum fetch buffer size to not exceed the configured memory limit. Take into account the number of partitions handled by the sources and inevitable pod outages. 
-  Set default pod memory request to 100MB, capacity of 50, so each vreplica memory limit is 2MB. With these parameters, the maximum number of partitions a source can handle to process 64k messages is about 16.  
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **lionelvillard** after the PR has been reviewed.

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/main/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":["lionelvillard"]} -->
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/source/client/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1371914709124517888/artifacts/line-cov.html#file1) | 48.0% | 41.4% | -6.6

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 802ad584519aee8bb8058c134ce6f348df37d21b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-unit-tests/1371914709095157760) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/470). [Your PR dashboard](https://gubernator.knative.dev/pr/lionelvillard).

<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 -->
knative-prow-robot: @lionelvillard: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 802ad584519aee8bb8058c134ce6f348df37d21b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-unit-tests/1371914709095157760) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | 802ad584519aee8bb8058c134ce6f348df37d21b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1371914709124517888) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/470). [Your PR dashboard](https://gubernator.knative.dev/pr/lionelvillard).

<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 -->
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/source/client/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1371921105371009024/artifacts/line-cov.html#file1) | 48.0% | 41.4% | -6.6
[pkg/source/mtadapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1371921105371009024/artifacts/line-cov.html#file2) | 68.4% | 48.3% | -20.1

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | c1af748ecd7e6a795bd58d2c88e5a1e0def465d7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1371921105371009024) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/470). [Your PR dashboard](https://gubernator.knative.dev/pr/lionelvillard).

<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 -->
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/source/client/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1371928134013685760/artifacts/line-cov.html#file1) | 48.0% | 41.4% | -6.6
[pkg/source/mtadapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1371928134013685760/artifacts/line-cov.html#file2) | 68.4% | 48.3% | -20.1

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | d6cfda30a4ae4beae5aa37c7c4e7304b4e5b9d0a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1371928134013685760) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/470). [Your PR dashboard](https://gubernator.knative.dev/pr/lionelvillard).

<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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/470?src=pr&el=h1) Report
> Merging [#470](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/470?src=pr&el=desc) (c1af748) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/cca17c800a6bf249e369a05974119798a837a987?el=desc) (cca17c8) will **decrease** coverage by `0.43%`.
> The diff coverage is `44.28%`.

> :exclamation: Current head c1af748 differs from pull request most recent head d6cfda3. Consider uploading reports for the commit d6cfda3 to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/470/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/470?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##             main     #470      +/-   ##
==========================================
- Coverage   73.80%   73.37%   -0.44%     
==========================================
  Files         138      138              
  Lines        5326     5371      +45     
==========================================
+ Hits         3931     3941      +10     
- Misses       1148     1182      +34     
- Partials      247      248       +1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/470?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/source/client/client.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/470/diff?src=pr&el=tree#diff-cGtnL3NvdXJjZS9jbGllbnQvY2xpZW50Lmdv) | `35.13% <0.00%> (-4.26%)` | :arrow_down: |
| [pkg/source/mtadapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/470/diff?src=pr&el=tree#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvYWRhcHRlci5nbw==) | `52.89% <40.67%> (-17.23%)` | :arrow_down: |
| [pkg/common/scheduler/placement.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/470/diff?src=pr&el=tree#diff-cGtnL2NvbW1vbi9zY2hlZHVsZXIvcGxhY2VtZW50Lmdv) | `100.00% <100.00%> (ø)` | |

------

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

knative-prow-robot: @lionelvillard: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | d6cfda30a4ae4beae5aa37c7c4e7304b4e5b9d0a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1371928134013685760) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-upgrade-tests | d6cfda30a4ae4beae5aa37c7c4e7304b4e5b9d0a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-upgrade-tests/1379810633435844608) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/470). [Your PR dashboard](https://gubernator.knative.dev/pr/lionelvillard).

<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 -->
knative-prow-robot: @lionelvillard: PR needs rebase.

<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>
knative-prow-robot: @lionelvillard: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-upgrade-tests | d6cfda30a4ae4beae5aa37c7c4e7304b4e5b9d0a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-upgrade-tests/1379810633435844608) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | 053758915d018239b0f453f09b2e0fbfb8466e0a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1380262347372761088) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/470). [Your PR dashboard](https://gubernator.knative.dev/pr/lionelvillard).

<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 -->
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/source/client/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1380268333827362816/artifacts/line-cov.html#file1) | 48.0% | 41.4% | -6.6
[pkg/source/mtadapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1380268333827362816/artifacts/line-cov.html#file2) | 68.4% | 48.3% | -20.1

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | c592f00202f9ead61e10bc96fad62875d90f4821 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1380268333827362816) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/470). [Your PR dashboard](https://gubernator.knative.dev/pr/lionelvillard).

<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 -->
knative-prow-robot: @lionelvillard: PR needs rebase.

<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>
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/470#pullrequestreview-635179524" title="Approved">steven0711dong</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **lionelvillard** after the PR has been reviewed.

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/main/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":["lionelvillard"]} -->
steven0711dong: /lgtm
knative-prow-robot: New changes are detected. LGTM label has been removed.
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **lionelvillard** after the PR has been reviewed.

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/main/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":["lionelvillard"]} -->
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/source/client/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1382317639799410688/artifacts/line-cov.html#file1) | 48.0% | 41.4% | -6.6
[pkg/source/mtadapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1382317639799410688/artifacts/line-cov.html#file2) | 68.4% | 48.3% | -20.1

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | 2f19e1f2e26a9de23085ba853cc565dc1a79b1f0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/470/pull-knative-sandbox-eventing-kafka-go-coverage/1382317639799410688) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/470). [Your PR dashboard](https://gubernator.knative.dev/pr/lionelvillard).

<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 -->
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/470#pullrequestreview-636084167" title="Approved">steven0711dong</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **lionelvillard** after the PR has been reviewed.

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/main/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":["lionelvillard"]} -->
steven0711dong: /lgtm

Patch:
diff --git a/config/source/multi/deployments/adapter.yaml b/config/source/multi/deployments/adapter.yaml
index 9f96e7767..9cebb7c89 100644
--- a/config/source/multi/deployments/adapter.yaml
+++ b/config/source/multi/deployments/adapter.yaml
@@ -43,6 +43,17 @@ spec:
             fieldRef:
               fieldPath: metadata.name
 
+        # The maximum number of messages per second, per vreplica
+        - name: VREPLICA_LIMITS_MPS
+          value: '50'
+
+        # The memory limit, per vreplica. Must be a quantity.
+        # see https://github.com/kubernetes/apimachinery/blob/master/pkg/api/resource/quantity.go#L31
+        # Should be pod requested memory / pod capacity (see controller.yaml)
+        - name: VREPLICA_LIMITS_MEMORY
+          value: '2Mi'
+
+
         # DO NOT MODIFY: The values below are being filled by the kafka source controller
         # See 500-controller.yaml
         - name: K_METRICS_CONFIG
diff --git a/config/source/multi/deployments/controller.yaml b/config/source/multi/deployments/controller.yaml
index 8ef720d7a..f3c6f6f06 100644
--- a/config/source/multi/deployments/controller.yaml
+++ b/config/source/multi/deployments/controller.yaml
@@ -45,11 +45,11 @@ spec:
         - name: CONFIG_LEADERELECTION_NAME
           value: config-leader-election
 
-        # how often (in seconds) the autoscaler tries to scale down the statefulset
+        # How often (in seconds) the autoscaler tries to scale down the statefulset.
         - name: AUTOSCALER_REFRESH_PERIOD
           value: '10'
 
-        # the number of virtual replicas this pod can handle
+        # The number of virtual replicas this pod can handle.
         - name: POD_CAPACITY
           value: '100'
 
diff --git a/pkg/common/scheduler/placement.go b/pkg/common/scheduler/placement.go
index ec47f8d0c..86dd89f4a 100644
--- a/pkg/common/scheduler/placement.go
+++ b/pkg/common/scheduler/placement.go
@@ -17,6 +17,7 @@ limitations under the License.
 package scheduler
 
 import (
+	"k8s.io/apimachinery/pkg/util/sets"
 	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
 )
 
@@ -38,3 +39,14 @@ func GetPlacementForPod(placements []duckv1alpha1.Placement, podName string) *du
 	}
 	return nil
 }
+
+// GetPodCount returns the number of pods with the given placements
+func GetPodCount(placements []duckv1alpha1.Placement) int {
+	set := sets.NewString()
+	for _, p := range placements {
+		if p.VReplicas > 0 {
+			set.Insert(p.PodName)
+		}
+	}
+	return set.Len()
+}
diff --git a/pkg/source/client/client.go b/pkg/source/client/client.go
index f07779430..9b5b04f31 100644
--- a/pkg/source/client/client.go
+++ b/pkg/source/client/client.go
@@ -119,3 +119,12 @@ func NewProducer(ctx context.Context) (sarama.Client, error) {
 
 	return sarama.NewClient(bs, cfg)
 }
+
+// NewProducer is a helper method for constructing an admin client
+func MakeAdminClient(ctx context.Context, env *KafkaEnvConfig) (sarama.ClusterAdmin, error) {
+	bs, cfg, err := NewConfigWithEnv(ctx, env)
+	if err != nil {
+		return nil, err
+	}
+	return sarama.NewClusterAdmin(bs, cfg)
+}
diff --git a/pkg/source/mtadapter/adapter.go b/pkg/source/mtadapter/adapter.go
index 1518a3684..bcd87683a 100644
--- a/pkg/source/mtadapter/adapter.go
+++ b/pkg/source/mtadapter/adapter.go
@@ -18,12 +18,15 @@ package mtadapter
 
 import (
 	"context"
+	"math"
+	"strconv"
 	"sync"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"go.uber.org/zap"
 	"golang.org/x/time/rate"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 
@@ -42,7 +45,9 @@ import (
 type AdapterConfig struct {
 	adapter.EnvConfig
 
-	PodName string `envconfig:"POD_NAME" required:"true"`
+	PodName     string `envconfig:"POD_NAME" required:"true"`
+	MPSLimit    int    `envconfig:"VREPLICA_LIMITS_MPS" required:"true"`
+	MemoryLimit string `envconfig:"VREPLICA_LIMITS_MEMORY" required:"true"`
 }
 
 func NewEnvConfig() adapter.EnvConfigAccessor {
@@ -55,6 +60,7 @@ type Adapter struct {
 	client      cloudevents.Client
 	adapterCtor adapter.MessageAdapterConstructor
 	kubeClient  kubernetes.Interface
+	memLimit    int64
 
 	sourcesMu sync.RWMutex
 	sources   map[string]context.CancelFunc
@@ -69,6 +75,7 @@ func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ceClie
 func newAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ceClient cloudevents.Client, adapterCtor adapter.MessageAdapterConstructor) adapter.Adapter {
 	logger := logging.FromContext(ctx)
 	config := processed.(*AdapterConfig)
+	ml := resource.MustParse(config.MemoryLimit)
 
 	return &Adapter{
 		client:      ceClient,
@@ -76,6 +83,7 @@ func newAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ceClie
 		logger:      logger,
 		adapterCtor: adapterCtor,
 		kubeClient:  kubeclient.Get(ctx),
+		memLimit:    ml.Value(),
 		sourcesMu:   sync.RWMutex{},
 		sources:     make(map[string]context.CancelFunc),
 	}
@@ -87,52 +95,124 @@ func (a *Adapter) Start(ctx context.Context) error {
 	return nil
 }
 
+// bufferSize returns the maximum fetch buffer size (in bytes)
+// so that the st adapter memory consumption does not exceed
+// the allocated memory per vreplica (see MemoryLimit).
+// Account pod (consumer) partial outage.
+func (a *Adapter) bufferSize(ctx context.Context,
+	logger *zap.SugaredLogger,
+	kafkaEnvConfig *client.KafkaEnvConfig,
+	topics []string,
+	podCount int) (int, error) {
+
+	// Compute the number of partitions handled by this source
+	// TODO: periodically check for # of resources. Need control-protocol.
+	adminClient, err := client.MakeAdminClient(ctx, kafkaEnvConfig)
+	if err != nil {
+		logger.Errorw("cannot create admin client", zap.Error(err))
+		return 0, err
+	}
+
+	metas, err := adminClient.DescribeTopics(topics)
+	if err != nil {
+		logger.Errorw("cannot describe topics", zap.Error(err))
+		return 0, err
+	}
+
+	totalPartitions := 0
+	for _, meta := range metas {
+		totalPartitions += len(meta.Partitions)
+	}
+	adminClient.Close()
+
+	partitionsPerPod := int(math.Ceil(float64(totalPartitions) / float64(podCount)))
+
+	// Ideally, partitions are evenly spread across Kafka consumers.
+	// However, due to rebalancing or consumer (un)availability, a consumer
+	// might have to handle more partitions than expected.
+	// For now, account for 1 unavailable consumer.
+	handledPartitions := 2 * partitionsPerPod
+	if podCount < 3 {
+		handledPartitions = totalPartitions
+	}
+
+	logger.Infow("partition count",
+		zap.Int("total", totalPartitions),
+		zap.Int("averagePerPod", partitionsPerPod),
+		zap.Int("handled", handledPartitions))
+
+	// A partition consumes about 2 * fetch buffer size.
+	return int(math.Floor(float64(a.memLimit) / float64(handledPartitions) / 2.0)), nil
+}
+
 // Implements MTAdapter
 
 func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 	a.sourcesMu.Lock()
 	defer a.sourcesMu.Unlock()
-	a.logger.Infow("adding source", "name", obj.Name)
 
 	key := obj.Namespace + "/" + obj.Name
 
+	logger := a.logger.With("key", key)
+	logger.Info("updating source")
+
 	cancel, ok := a.sources[key]
 
 	if ok {
 		// TODO: do not stop if the only thing that changes is the number of vreplicas
-		a.logger.Info("stopping adapter", zap.String("key", key))
+		logger.Info("stopping adapter")
 		cancel()
 	}
 
 	placement := scheduler.GetPlacementForPod(obj.GetPlacements(), a.config.PodName)
 	if placement == nil || placement.VReplicas == 0 {
 		// this pod does not handle this source. Skipping
-		a.logger.Infow("no replicas assigned to this source. skipping", zap.String("key", key))
+		logger.Info("no replicas assigned to this source. skipping")
 		return
 	}
 
+	kafkaEnvConfig := client.KafkaEnvConfig{
+		BootstrapServers: obj.Spec.BootstrapServers,
+		Net: client.AdapterNet{
+			SASL: client.AdapterSASL{
+				Enable:   obj.Spec.Net.SASL.Enable,
+				User:     a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.User.SecretKeyRef),
+				Password: a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.Password.SecretKeyRef),
+				Type:     a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.Type.SecretKeyRef),
+			},
+			TLS: client.AdapterTLS{
+				Enable: obj.Spec.Net.TLS.Enable,
+				Cert:   a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.Cert.SecretKeyRef),
+				Key:    a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.Key.SecretKeyRef),
+				CACert: a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.CACert.SecretKeyRef),
+			},
+		},
+	}
+
+	// Enforce memory limits
+	if a.memLimit > 0 {
+		// TODO: periodically enforce limits as the number of partitions can dynamically change
+		bufferSizePerVReplica, err := a.bufferSize(ctx, logger, &kafkaEnvConfig, obj.Spec.Topics, scheduler.GetPodCount(obj.Status.Placement))
+		if err != nil {
+			return
+		}
+		bufferSize := bufferSizePerVReplica * int(placement.VReplicas)
+		a.logger.Infow("setting fetch buffer size", zap.Int("size", bufferSize))
+
+		// Nasty.
+		bufferSizeStr := strconv.Itoa(bufferSize)
+		min := `\n    Min: ` + bufferSizeStr
+		def := `\n    Default: ` + bufferSizeStr
+		max := `\n    Max: ` + bufferSizeStr
+		kafkaEnvConfig.KafkaConfigJson = `{"sarama": "Consumer:\n  Fetch:` + min + def + max + `"}`
+	}
+
 	config := stadapter.AdapterConfig{
 		EnvConfig: adapter.EnvConfig{
 			Component: "kafkasource",
 			Namespace: obj.Namespace,
 		},
-		KafkaEnvConfig: client.KafkaEnvConfig{
-			BootstrapServers: obj.Spec.BootstrapServers,
-			Net: client.AdapterNet{
-				SASL: client.AdapterSASL{
-					Enable:   obj.Spec.Net.SASL.Enable,
-					User:     a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.User.SecretKeyRef),
-					Password: a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.Password.SecretKeyRef),
-					Type:     a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.Type.SecretKeyRef),
-				},
-				TLS: client.AdapterTLS{
-					Enable: obj.Spec.Net.TLS.Enable,
-					Cert:   a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.Cert.SecretKeyRef),
-					Key:    a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.Key.SecretKeyRef),
-					CACert: a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.CACert.SecretKeyRef),
-				},
-			},
-		},
+		KafkaEnvConfig:       kafkaEnvConfig,
 		Topics:               obj.Spec.Topics,
 		ConsumerGroup:        obj.Spec.ConsumerGroup,
 		Name:                 obj.Name,
@@ -155,9 +235,9 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 
 	adapter := a.adapterCtor(ctx, &config, httpBindingsSender, reporter)
 
-	if a, ok := adapter.(*stadapter.Adapter); ok {
-		// TODO: configurable
-		a.SetRateLimits(rate.Limit(10.0*placement.VReplicas), 20*int(placement.VReplicas))
+	// TODO: define Limit interface.
+	if sta, ok := adapter.(*stadapter.Adapter); ok {
+		sta.SetRateLimits(rate.Limit(a.config.MPSLimit*int(placement.VReplicas)), 2*a.config.MPSLimit*int(placement.VReplicas))
 	}
 
 	ctx, cancelFn := context.WithCancel(ctx)
diff --git a/pkg/source/reconciler/mtsource/controller.go b/pkg/source/reconciler/mtsource/controller.go
index 85025fccb..7c6ccbb93 100644
--- a/pkg/source/reconciler/mtsource/controller.go
+++ b/pkg/source/reconciler/mtsource/controller.go
@@ -70,7 +70,6 @@ func NewController(
 	sourcesv1beta1.RegisterAlternateKafkaConditionSet(sourcesv1beta1.KafkaMTSourceCondSet)
 
 	rp := time.Duration(env.SchedulerRefreshPeriod) * time.Second
-
 	c.scheduler = scheduler.NewScheduler(ctx, system.Namespace(), mtadapterName, c.vpodLister, rp, env.PodCapacity)
 
 	logging.FromContext(ctx).Info("Setting up kafka event handlers")


Test patch:
diff --git a/pkg/common/scheduler/placement_test.go b/pkg/common/scheduler/placement_test.go
index 713c187b5..9534b13e3 100644
--- a/pkg/common/scheduler/placement_test.go
+++ b/pkg/common/scheduler/placement_test.go
@@ -106,3 +106,51 @@ func TestGetPlacementForPod(t *testing.T) {
 		})
 	}
 }
+func TestPodCount(t *testing.T) {
+	testCases := []struct {
+		name       string
+		placements []duckv1alpha1.Placement
+		expected   int
+	}{
+		{
+			name:       "nil placements",
+			placements: nil,
+			expected:   0,
+		},
+		{
+			name:       "empty placements",
+			placements: []duckv1alpha1.Placement{},
+			expected:   0,
+		},
+		{
+			name:       "one pod",
+			placements: []duckv1alpha1.Placement{{PodName: "d", VReplicas: 2}},
+			expected:   1,
+		},
+		{
+			name: "two pods",
+			placements: []duckv1alpha1.Placement{
+				{PodName: "p1", VReplicas: 2},
+				{PodName: "p2", VReplicas: 6},
+				{PodName: "p1", VReplicas: 6}},
+			expected: 2,
+		},
+		{
+			name: "three pods, one with no vreplicas",
+			placements: []duckv1alpha1.Placement{
+				{PodName: "p1", VReplicas: 2},
+				{PodName: "p2", VReplicas: 6},
+				{PodName: "p1", VReplicas: 0}},
+			expected: 2,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := GetPodCount(tc.placements)
+			if got != tc.expected {
+				t.Errorf("got %v, want %v", got, tc.expected)
+			}
+		})
+	}
+}
diff --git a/pkg/source/mtadapter/adapter_test.go b/pkg/source/mtadapter/adapter_test.go
index 9a7cd3f9e..f7bf2231b 100644
--- a/pkg/source/mtadapter/adapter_test.go
+++ b/pkg/source/mtadapter/adapter_test.go
@@ -21,8 +21,6 @@ import (
 	"testing"
 	"time"
 
-	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
-
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	_ "knative.dev/pkg/client/injection/kube/client/fake"
@@ -34,6 +32,7 @@ import (
 	"knative.dev/eventing/pkg/kncloudevents"
 
 	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 )
 
@@ -50,7 +49,7 @@ func TestUpdateRemoveSources(t *testing.T) {
 	ctx, _ := pkgtesting.SetupFakeContext(t)
 	ctx, cancelAdapter := context.WithCancel(ctx)
 
-	env := &AdapterConfig{PodName: podName}
+	env := &AdapterConfig{PodName: podName, MemoryLimit: "0"}
 	ceClient := adaptertest.NewTestClient()
 
 	adapter := newAdapter(ctx, env, ceClient, newSampleAdapter).(*Adapter)