Repository: knative-sandbox/eventing-kafka
Pull request #668 (2021-05-28 14:07:29)

Adjust MaxResponseSize based on the number of active stadapters
<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- MaxResponseSize is now Container Memory request / # of source instances in container. Cap at 50MB 
-
-

<!--
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 **APPROVED**

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

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

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

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

- ~~[config/source/multi/OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/main/config/source/multi/OWNERS)~~ [lionelvillard]
- ~~[pkg/source/OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/main/pkg/source/OWNERS)~~ [lionelvillard]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
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/mtadapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/668/pull-knative-sandbox-eventing-kafka-go-coverage/1398279775251861504/artifacts/line-cov.html#file0) | 75.0% | 77.0% | 2.0

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/668?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#668](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/668?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (6af6777) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/a7e5fa6213d39ed8ac77efc2073c530fcdd21913?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (a7e5fa6) will **increase** coverage by `0.04%`.
> The diff coverage is `85.71%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/668/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/668?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@            Coverage Diff             @@
##             main     #668      +/-   ##
==========================================
+ Coverage   74.43%   74.47%   +0.04%     
==========================================
  Files         127      127              
  Lines        5534     5543       +9     
==========================================
+ Hits         4119     4128       +9     
  Misses       1198     1198              
  Partials      217      217              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/668?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [pkg/source/mtadapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/668/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvYWRhcHRlci5nbw==) | `72.93% <85.71%> (+1.96%)` | :arrow_up: |

------

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

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/mtadapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/668/pull-knative-sandbox-eventing-kafka-go-coverage/1398291443985944576/artifacts/line-cov.html#file0) | 75.0% | 78.0% | 3.0

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/source/mtadapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/668/pull-knative-sandbox-eventing-kafka-go-coverage/1398302701015535616/artifacts/line-cov.html#file0) | 75.0% | 78.2% | 3.2

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/mtadapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/668/pull-knative-sandbox-eventing-kafka-go-coverage/1399714372976119808/artifacts/line-cov.html#file0) | 75.0% | 78.2% | 3.2

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/668#" title="Author self-approved">lionelvillard</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/668#pullrequestreview-673174565" title="Approved">steven0711dong</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:

- ~~[config/source/multi/OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/main/config/source/multi/OWNERS)~~ [lionelvillard,steven0711dong]
- ~~[pkg/source/OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/main/pkg/source/OWNERS)~~ [lionelvillard,steven0711dong]

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

Patch:
diff --git a/config/source/multi/deployments/adapter.yaml b/config/source/multi/deployments/adapter.yaml
index 8822f7faa..fd1e7bc4b 100644
--- a/config/source/multi/deployments/adapter.yaml
+++ b/config/source/multi/deployments/adapter.yaml
@@ -47,12 +47,9 @@ spec:
         - 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'
-
+        # The pod requested memory (see resources.requests.memory)
+        - name: REQUESTS_MEMORY
+          value: '300Mi'
 
         # DO NOT MODIFY: The values below are being filled by the kafka source controller
         # See 500-controller.yaml
@@ -68,10 +65,10 @@ spec:
         resources:
           requests:
             cpu: 100m
-            memory: 250Mi
+            memory: 300Mi
           limits:
             cpu: 200m
-            memory: 300Mi
+            memory: 500Mi
 
         ports:
         - name: metrics
diff --git a/config/source/multi/deployments/controller.yaml b/config/source/multi/deployments/controller.yaml
index 431160d9a..e4bf9a5db 100644
--- a/config/source/multi/deployments/controller.yaml
+++ b/config/source/multi/deployments/controller.yaml
@@ -49,7 +49,7 @@ spec:
         - name: AUTOSCALER_REFRESH_PERIOD
           value: '100'
 
-        # The number of virtual replicas this pod can handle.
+        # The number of virtual replicas each adapter pod can handle.
         - name: POD_CAPACITY
           value: '100'
 
diff --git a/pkg/source/mtadapter/adapter.go b/pkg/source/mtadapter/adapter.go
index 06c31528c..5aa7007c7 100644
--- a/pkg/source/mtadapter/adapter.go
+++ b/pkg/source/mtadapter/adapter.go
@@ -42,12 +42,16 @@ import (
 	"knative.dev/eventing-kafka/pkg/source/client"
 )
 
+const (
+	responseSizeCap = 50 * 1024 * 1024
+)
+
 type AdapterConfig struct {
 	adapter.EnvConfig
 
-	PodName     string `envconfig:"POD_NAME" required:"true"`
-	MPSLimit    int    `envconfig:"VREPLICA_LIMITS_MPS" required:"true"`
-	MemoryLimit string `envconfig:"VREPLICA_LIMITS_MEMORY" required:"true"`
+	PodName       string `envconfig:"POD_NAME" required:"true"`
+	MPSLimit      int    `envconfig:"VREPLICA_LIMITS_MPS" required:"true"`
+	MemoryRequest string `envconfig:"REQUESTS_MEMORY" required:"true"`
 }
 
 func NewEnvConfig() adapter.EnvConfigAccessor {
@@ -65,7 +69,8 @@ type Adapter struct {
 	client      cloudevents.Client
 	adapterCtor adapter.MessageAdapterConstructor
 	kubeClient  kubernetes.Interface
-	memLimit    int32
+
+	memoryRequest int64
 
 	sourcesMu sync.RWMutex
 	sources   map[string]cancelContext
@@ -80,17 +85,19 @@ 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)
 
+	sarama.MaxResponseSize = responseSizeCap
+
+	mr := resource.MustParse(config.MemoryRequest)
 	return &Adapter{
-		client:      ceClient,
-		config:      config,
-		logger:      logger,
-		adapterCtor: adapterCtor,
-		kubeClient:  kubeclient.Get(ctx),
-		memLimit:    int32(ml.Value()),
-		sourcesMu:   sync.RWMutex{},
-		sources:     make(map[string]cancelContext),
+		client:        ceClient,
+		config:        config,
+		logger:        logger,
+		adapterCtor:   adapterCtor,
+		kubeClient:    kubeclient.Get(ctx),
+		memoryRequest: mr.Value(),
+		sourcesMu:     sync.RWMutex{},
+		sources:       make(map[string]cancelContext),
 	}
 }
 
@@ -123,6 +130,7 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) error {
 
 		// Nothing to stop anymore
 		delete(a.sources, key)
+		a.adjustResponseSize()
 	}
 
 	placement := scheduler.GetPlacementForPod(obj.GetPlacements(), a.config.PodName)
@@ -215,11 +223,6 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) error {
 		sta.SetRateLimits(rate.Limit(a.config.MPSLimit*int(placement.VReplicas)), 2*a.config.MPSLimit*int(placement.VReplicas))
 	}
 
-	if a.memLimit > 0 {
-		sarama.MaxResponseSize = placement.VReplicas * a.memLimit
-		logger.Infof("Setting MaxResponseSize to %d bytes", sarama.MaxResponseSize)
-	}
-
 	ctx, cancelFn := context.WithCancel(ctx)
 
 	cancel = cancelContext{
@@ -227,6 +230,9 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) error {
 		stopped: make(chan bool),
 	}
 
+	a.sources[key] = cancel
+	a.adjustResponseSize()
+
 	go func(ctx context.Context) {
 		err := adapter.Start(ctx)
 		if err != nil {
@@ -235,7 +241,6 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) error {
 		cancel.stopped <- true
 	}(ctx)
 
-	a.sources[key] = cancel
 	a.logger.Infow("source added", "name", obj.Name)
 	return nil
 }
@@ -258,6 +263,8 @@ func (a *Adapter) Remove(obj *v1beta1.KafkaSource) {
 	<-cancel.stopped
 
 	delete(a.sources, key)
+	a.adjustResponseSize()
+
 	a.logger.Infow("source removed", "name", obj.Name, "remaining", len(a.sources))
 }
 
@@ -279,3 +286,24 @@ func (a *Adapter) ResolveSecret(ctx context.Context, ns string, ref *corev1.Secr
 	a.logger.Errorw("missing secret key or empty secret value", zap.String("secretname", ref.Name), zap.String("secretkey", ref.Key))
 	return "", fmt.Errorf("missing secret key or empty secret value (%s/%s)", ref.Name, ref.Key)
 }
+
+// adjustResponseSize ensures the sum of all Kafka clients memory usage does not exceed the container memory request.
+func (a *Adapter) adjustResponseSize() {
+	if a.memoryRequest > 0 {
+		maxResponseSize := int32(float64(a.memoryRequest) / float64(len(a.sources)))
+
+		// cap the response size to 50MB.
+		if maxResponseSize > responseSizeCap {
+			maxResponseSize = 50 * 1024 * 1024
+		}
+		// Check for compliance.
+		if maxResponseSize < 64*1024 {
+			// Not CloudEvent compliant (https://github.com/cloudevents/spec/blob/v1.0.1/spec.md#size-limits)
+			a.logger.Warnw("Kafka response size is lower than 64KB. Increase the pod memory request and/or lower the pod capacity.",
+				zap.Int32("responseSize", maxResponseSize))
+		}
+
+		sarama.MaxResponseSize = maxResponseSize
+		a.logger.Infof("Setting MaxResponseSize to %d bytes", sarama.MaxResponseSize)
+	}
+}


Test patch:
diff --git a/pkg/source/mtadapter/adapter_test.go b/pkg/source/mtadapter/adapter_test.go
index f29e6025f..499013ba7 100644
--- a/pkg/source/mtadapter/adapter_test.go
+++ b/pkg/source/mtadapter/adapter_test.go
@@ -18,14 +18,17 @@ package mtadapter
 
 import (
 	"context"
+	"strconv"
 	"testing"
 	"time"
 
+	"github.com/Shopify/sarama"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 
 	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	logtesting "knative.dev/pkg/logging/testing"
 	pkgtesting "knative.dev/pkg/reconciler/testing"
 	"knative.dev/pkg/source"
 
@@ -51,7 +54,7 @@ func TestUpdateRemoveSources(t *testing.T) {
 	ctx, _ := pkgtesting.SetupFakeContext(t)
 	ctx, cancelAdapter := context.WithCancel(ctx)
 
-	env := &AdapterConfig{PodName: podName, MemoryLimit: "0"}
+	env := &AdapterConfig{PodName: podName, MemoryRequest: "0"}
 	ceClient := adaptertest.NewTestClient()
 
 	mtadapter := newAdapter(ctx, env, ceClient, newSampleAdapter).(*Adapter)
@@ -352,7 +355,7 @@ func TestSourceMTAdapter(t *testing.T) {
 
 			ctx, cancelAdapter := context.WithCancel(ctx)
 
-			env := &AdapterConfig{PodName: podName, MemoryLimit: "0"}
+			env := &AdapterConfig{PodName: podName, MemoryRequest: "0"}
 			ceClient := adaptertest.NewTestClient()
 
 			adapter := newAdapter(ctx, env, ceClient, newSampleAdapter).(*Adapter)
@@ -405,3 +408,40 @@ func (d *sampleAdapter) Start(ctx context.Context) error {
 
 	return nil
 }
+
+func TestAdjustResponseSize(t *testing.T) {
+	testCases := map[string]struct {
+		memoryRequest int64
+		sourceCount   int
+		want          int32
+	}{
+		"no memory request":                           {memoryRequest: 0, sourceCount: 1, want: 100 * 1024 * 1024},
+		"memory request, response size less 64k":      {memoryRequest: 128 * 1024, sourceCount: 4, want: 32 * 1024},
+		"memory request, response size more 64k":      {memoryRequest: 512 * 1024, sourceCount: 4, want: 128 * 1024},
+		"memory request, response size more than cap": {memoryRequest: 100 * 1024 * 1024, sourceCount: 1, want: responseSizeCap},
+	}
+
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			// must run sequentially.
+			sarama.MaxResponseSize = 100 * 1024 * 1024
+
+			sources := make(map[string]cancelContext)
+			for i := 0; i < tc.sourceCount; i++ {
+				sources["s"+strconv.Itoa(i)] = cancelContext{}
+			}
+
+			a := Adapter{
+				memoryRequest: tc.memoryRequest,
+				sources:       sources,
+				logger:        logtesting.TestLogger(t),
+			}
+			a.adjustResponseSize()
+
+			if sarama.MaxResponseSize != tc.want {
+				t.Errorf("Unexpected MaxResponseSize. wanted %d, got %d", tc.want, sarama.MaxResponseSize)
+			}
+		})
+	}
+
+}