Repository: knative-sandbox/eventing-kafka (go)

Issue:
[mtsource] mtadapter crashes due to OOM
**Describe the bug**
When the mtadapter starts it sometimes enters in the crashloop due to OOM. This happens when there are many queued messages in a topic, leading to the input buffers to exceed the container memory limit. 

MaxResponseSize was used to mitigate this issue but its scope is per Fetch Request, which when the consumer group starts, the number of Fetch Requests can somewhere between 1 to number of partitions, until all partitions are assigned to a broker, in which case the number of Fetch requests stabilizes to the number of brokers. 


 **Expected behavior**
A clear and concise description of what you expected to happen.

No crash

 **To Reproduce**
Steps to reproduce the behavior.
Create a KafkaSource, Produce event to a topic. Delete the KafkaSource and wait some time for the topic to be filled with messages. Recreate the KafkaSource, using the same consumergroup and observe the crash.

 **Knative release version**

 **Additional context**
Add any other context about the problem here such as proposed priority


Pull request description:
Revert to setting consumer.fetch options instead of MaxReponseSize
Fixes #691

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Revert https://github.com/knative-sandbox/eventing-kafka/pull/668/files as MaxResponseSize does not work as expected (see #691 for an explanation)
- Beef up the adapter pod to be able to handle larger messages. Not a definitive number, we can adjust as we gather more data.
-

<!--
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.
-->


Patch:
diff --git a/config/source/multi/deployments/adapter.yaml b/config/source/multi/deployments/adapter.yaml
index fd1e7bc4b..ec1936dae 100644
--- a/config/source/multi/deployments/adapter.yaml
+++ b/config/source/multi/deployments/adapter.yaml
@@ -47,9 +47,11 @@ spec:
         - name: VREPLICA_LIMITS_MPS
           value: '50'
 
-        # The pod requested memory (see resources.requests.memory)
-        - name: REQUESTS_MEMORY
-          value: '300Mi'
+        # 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 - overhead) / pod capacity (see controller.yaml)
+        - name: VREPLICA_LIMITS_MEMORY
+          value: '6Mi'
 
         # DO NOT MODIFY: The values below are being filled by the kafka source controller
         # See 500-controller.yaml
@@ -64,11 +66,11 @@ spec:
 
         resources:
           requests:
-            cpu: 100m
-            memory: 300Mi
+            cpu: 1000m
+            memory: 700Mi # 600Mi for vreplicas + 100Mi overhead
           limits:
-            cpu: 200m
-            memory: 500Mi
+            cpu: 2000m
+            memory: 1000Mi
 
         ports:
         - name: metrics
diff --git a/go.mod b/go.mod
index 5150768eb..dfb125e04 100644
--- a/go.mod
+++ b/go.mod
@@ -35,7 +35,7 @@ require (
 	k8s.io/client-go v0.19.7
 	k8s.io/utils v0.0.0-20200729134348-d5654de09c73
 	knative.dev/control-protocol v0.0.0-20210604132745-1e48321b7e9b
-	knative.dev/eventing v0.23.1-0.20210608085741-b05d993d260e
+	knative.dev/eventing v0.23.1-0.20210608130141-92d85bf916e9
 	knative.dev/hack v0.0.0-20210601210329-de04b70e00d0
 	knative.dev/networking v0.0.0-20210603073844-5521a8b92648
 	knative.dev/pkg v0.0.0-20210602095030-0e61d6763dd6
diff --git a/go.sum b/go.sum
index 1caea9162..c5aa817c8 100644
--- a/go.sum
+++ b/go.sum
@@ -1311,8 +1311,8 @@ k8s.io/utils v0.0.0-20200729134348-d5654de09c73 h1:uJmqzgNWG7XyClnU/mLPBWwfKKF1K
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20210604132745-1e48321b7e9b h1:LKpJLida2PnUhSHZ1Eu+kPHUDJEiy1WX5Uec075k43Y=
 knative.dev/control-protocol v0.0.0-20210604132745-1e48321b7e9b/go.mod h1:iRE6O4mY88gH6xWl6ZFhG6sJobzrZjFrz5xhX4h93Ns=
-knative.dev/eventing v0.23.1-0.20210608085741-b05d993d260e h1:ti5tC3CoSxRcjfEvKu11zM9/tZpfjiSGvqUAqj/XojU=
-knative.dev/eventing v0.23.1-0.20210608085741-b05d993d260e/go.mod h1:mYAQz8Jk3S5Rd24Drmp7hWX+U3JzTXUEvMmfI62VlOM=
+knative.dev/eventing v0.23.1-0.20210608130141-92d85bf916e9 h1:yo2PJ58hGS2PELz7fsSHNJsmcH2fzCwij0kRgk73vAA=
+knative.dev/eventing v0.23.1-0.20210608130141-92d85bf916e9/go.mod h1:mYAQz8Jk3S5Rd24Drmp7hWX+U3JzTXUEvMmfI62VlOM=
 knative.dev/hack v0.0.0-20210601210329-de04b70e00d0 h1:GXGtVDptFpq2aFTh/m+Xii0kvGIcGZQQtx/sHinZqZs=
 knative.dev/hack v0.0.0-20210601210329-de04b70e00d0/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210601210329-de04b70e00d0/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index 1884a6e56..634c2b137 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -67,9 +67,10 @@ func NewEnvConfig() adapter.EnvConfigAccessor {
 }
 
 type Adapter struct {
-	config        *AdapterConfig
-	controlServer *ctrlnetwork.ControlServer
-	saramaConfig  *sarama.Config
+	config            *AdapterConfig
+	controlServer     *ctrlnetwork.ControlServer
+	saramaConfig      *sarama.Config
+	offsetInitialized bool
 
 	httpMessageSender *kncloudevents.HTTPMessageSender
 	reporter          pkgsource.StatsReporter
@@ -242,7 +243,7 @@ func (a *Adapter) Cleanup(sess sarama.ConsumerGroupSession) {
 
 // InitOffsets makes sure all consumer group offsets are set.
 func (a *Adapter) InitOffsets(session sarama.ConsumerGroupSession) error {
-	if a.saramaConfig.Consumer.Offsets.Initial == sarama.OffsetNewest {
+	if !a.offsetInitialized && a.saramaConfig.Consumer.Offsets.Initial == sarama.OffsetNewest {
 		// We want to make sure that ALL consumer group offsets are set to avoid
 		// losing events in case the consumer group session is closed before at least one message is
 		// consumed from ALL partitions.
@@ -302,6 +303,8 @@ func (a *Adapter) InitOffsets(session sarama.ConsumerGroupSession) error {
 
 			a.logger.Infow("consumer group offsets committed", zap.String("consumergroup", a.config.ConsumerGroup))
 		}
+
+		a.offsetInitialized = true
 	}
 
 	// At this stage the KafkaSource instance is considered Ready (TODO: update KafkaSource status)
diff --git a/pkg/source/mtadapter/adapter.go b/pkg/source/mtadapter/adapter.go
index dba2aa0fe..187ec09b4 100644
--- a/pkg/source/mtadapter/adapter.go
+++ b/pkg/source/mtadapter/adapter.go
@@ -19,9 +19,10 @@ package mtadapter
 import (
 	"context"
 	"fmt"
+	"math"
+	"strconv"
 	"sync"
 
-	"github.com/Shopify/sarama"
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"go.uber.org/zap"
 	"golang.org/x/time/rate"
@@ -42,17 +43,12 @@ import (
 	"knative.dev/eventing-kafka/pkg/source/client"
 )
 
-const (
-	responseSizeCap = int32(50 * 1024 * 1024)
-	maxBrokers      = 6
-)
-
 type AdapterConfig struct {
 	adapter.EnvConfig
 
-	PodName       string `envconfig:"POD_NAME" required:"true"`
-	MPSLimit      int    `envconfig:"VREPLICA_LIMITS_MPS" required:"true"`
-	MemoryRequest string `envconfig:"REQUESTS_MEMORY" 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 {
@@ -70,8 +66,7 @@ type Adapter struct {
 	client      cloudevents.Client
 	adapterCtor adapter.MessageAdapterConstructor
 	kubeClient  kubernetes.Interface
-
-	memoryRequest int64
+	memLimit    int32
 
 	sourcesMu sync.RWMutex
 	sources   map[string]cancelContext
@@ -87,18 +82,16 @@ func newAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ceClie
 	logger := logging.FromContext(ctx)
 	config := processed.(*AdapterConfig)
 
-	sarama.MaxResponseSize = responseSizeCap
-
-	mr := resource.MustParse(config.MemoryRequest)
+	ml := resource.MustParse(config.MemoryLimit)
 	return &Adapter{
-		client:        ceClient,
-		config:        config,
-		logger:        logger,
-		adapterCtor:   adapterCtor,
-		kubeClient:    kubeclient.Get(ctx),
-		memoryRequest: mr.Value(),
-		sourcesMu:     sync.RWMutex{},
-		sources:       make(map[string]cancelContext),
+		client:      ceClient,
+		config:      config,
+		logger:      logger,
+		adapterCtor: adapterCtor,
+		kubeClient:  kubeclient.Get(ctx),
+		memLimit:    int32(ml.Value()),
+		sourcesMu:   sync.RWMutex{},
+		sources:     make(map[string]cancelContext),
 	}
 }
 
@@ -131,7 +124,6 @@ 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)
@@ -189,6 +181,30 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) error {
 		},
 	}
 
+	// Enforce memory limits
+	if a.memLimit > 0 {
+		// TODO: periodically enforce limits as the number of partitions can dynamically change
+		fetchSizePerVReplica, err := a.partitionFetchSize(ctx, logger, &kafkaEnvConfig, obj.Spec.Topics, scheduler.GetPodCount(obj.Status.Placement))
+		if err != nil {
+			return err
+		}
+		fetchSize := fetchSizePerVReplica * int(placement.VReplicas)
+
+		// Must handle at least 64k messages to the compliant with the CloudEvent spec
+		maxFetchSize := fetchSize
+		if fetchSize < 64*1024 {
+			maxFetchSize = 64 * 1024
+		}
+		a.logger.Infow("setting partition fetch sizes", zap.Int("min", fetchSize), zap.Int("default", fetchSize), zap.Int("max", maxFetchSize))
+
+		// TODO: find a better way to interact with the ST adapter.
+		bufferSizeStr := strconv.Itoa(fetchSize)
+		min := `\n    Min: ` + bufferSizeStr
+		def := `\n    Default: ` + bufferSizeStr
+		max := `\n    Max: ` + strconv.Itoa(maxFetchSize)
+		kafkaEnvConfig.KafkaConfigJson = `{"SaramaYamlString": "Consumer:\n  Fetch:` + min + def + max + `"}`
+	}
+
 	config := stadapter.AdapterConfig{
 		EnvConfig: adapter.EnvConfig{
 			Component: "kafkasource",
@@ -232,7 +248,6 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) error {
 	}
 
 	a.sources[key] = cancel
-	a.adjustResponseSize()
 
 	go func(ctx context.Context) {
 		err := adapter.Start(ctx)
@@ -264,7 +279,6 @@ 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))
 }
@@ -288,32 +302,55 @@ func (a *Adapter) ResolveSecret(ctx context.Context, ns string, ref *corev1.Secr
 	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 := responseSizeCap
+// partitionFetchSize determines what should be the default fetch size (in bytes)
+// so that the st adapter memory consumption does not exceed
+// the allocated memory per vreplica (see MemoryLimit).
+// Account for pod (consumer) partial outage by reducing the
+// partition buffer size
+func (a *Adapter) partitionFetchSize(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
+	}
 
-		if len(a.sources) > 0 {
-			maxResponseSize = int32(float32(a.memoryRequest) / float32(len(a.sources)))
-		}
-		// cap the response size to 50MB.
-		if maxResponseSize > responseSizeCap {
-			maxResponseSize = 50 * 1024 * 1024
-		}
+	metas, err := adminClient.DescribeTopics(topics)
+	if err != nil {
+		logger.Errorw("cannot describe topics", zap.Error(err))
+		return 0, err
+	}
 
-		// maxResponseSize is per connected brokers.
-		// For now cap the number of connected brokers to maxBrokers until a better solution
-		// comes along
-		maxResponseSize = int32(float32(maxResponseSize) / float32(maxBrokers))
+	totalPartitions := 0
+	for _, meta := range metas {
+		totalPartitions += len(meta.Partitions)
+	}
+	adminClient.Close()
 
-		// 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))
-		}
+	partitionsPerPod := int(math.Ceil(float64(totalPartitions) / float64(podCount)))
 
-		sarama.MaxResponseSize = maxResponseSize
-		a.logger.Infof("Setting MaxResponseSize to %d bytes", sarama.MaxResponseSize)
+	// 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 partition size
+	// Once by FetchResponse blocks and a second time when those blocks are converted to messages
+	// see https://github.com/Shopify/sarama/blob/83d633e6e4f71b402df5e9c53ad5c1c334b7065d/consumer.go#L649
+	return int(math.Floor(float64(a.memLimit) / float64(handledPartitions) / 2.0)), nil
 }
diff --git a/vendor/knative.dev/eventing/pkg/channel/fanout/fanout_message_handler.go b/vendor/knative.dev/eventing/pkg/channel/fanout/fanout_message_handler.go
index ceb0cbef6..fcb140983 100644
--- a/vendor/knative.dev/eventing/pkg/channel/fanout/fanout_message_handler.go
+++ b/vendor/knative.dev/eventing/pkg/channel/fanout/fanout_message_handler.go
@@ -226,7 +226,7 @@ func (f *FanoutMessageHandler) ServeHTTP(response nethttp.ResponseWriter, reques
 }
 
 // ParseDispatchResultAndReportMetric processes the dispatch result and records the related channel metrics with the appropriate context
-func ParseDispatchResultAndReportMetrics(result dispatchResult, reporter channel.StatsReporter, reportArgs channel.ReportArgs) error {
+func ParseDispatchResultAndReportMetrics(result DispatchResult, reporter channel.StatsReporter, reportArgs channel.ReportArgs) error {
 	if result.info != nil && result.info.Time > channel.NoDuration {
 		if result.info.ResponseCode > channel.NoResponse {
 			_ = reporter.ReportEventDispatchTime(&reportArgs, result.info.ResponseCode, result.info.Time)
@@ -245,20 +245,20 @@ func ParseDispatchResultAndReportMetrics(result dispatchResult, reporter channel
 
 // dispatch takes the event, fans it out to each subscription in subs. If all the fanned out
 // events return successfully, then return nil. Else, return an error.
-func (f *FanoutMessageHandler) dispatch(ctx context.Context, subs []Subscription, bufferedMessage binding.Message, additionalHeaders nethttp.Header) dispatchResult {
+func (f *FanoutMessageHandler) dispatch(ctx context.Context, subs []Subscription, bufferedMessage binding.Message, additionalHeaders nethttp.Header) DispatchResult {
 	// Bind the lifecycle of the buffered message to the number of subs
 	bufferedMessage = buffering.WithAcksBeforeFinish(bufferedMessage, len(subs))
 
-	errorCh := make(chan dispatchResult, len(subs))
+	errorCh := make(chan DispatchResult, len(subs))
 	for _, sub := range subs {
 		go func(s Subscription) {
 			dispatchedResultPerSub, err := f.makeFanoutRequest(ctx, bufferedMessage, additionalHeaders, s)
-			errorCh <- dispatchResult{err: err, info: dispatchedResultPerSub}
+			errorCh <- DispatchResult{err: err, info: dispatchedResultPerSub}
 		}(sub)
 	}
 
 	var totalDispatchTimeForFanout time.Duration = channel.NoDuration
-	dispatchResultForFanout := dispatchResult{
+	dispatchResultForFanout := DispatchResult{
 		info: &channel.DispatchExecutionInfo{
 			Time:         channel.NoDuration,
 			ResponseCode: channel.NoResponse,
@@ -307,7 +307,22 @@ func (f *FanoutMessageHandler) makeFanoutRequest(ctx context.Context, message bi
 	)
 }
 
-type dispatchResult struct {
+type DispatchResult struct {
 	err  error
 	info *channel.DispatchExecutionInfo
 }
+
+func (d DispatchResult) Error() error {
+	return d.err
+}
+
+func (d DispatchResult) Info() *channel.DispatchExecutionInfo {
+	return d.info
+}
+
+func NewDispatchResult(err error, info *channel.DispatchExecutionInfo) DispatchResult {
+	return DispatchResult{
+		err:  err,
+		info: info,
+	}
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 675b1b52d..3bb3b047a 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1106,7 +1106,7 @@ knative.dev/control-protocol/pkg/message
 knative.dev/control-protocol/pkg/network
 knative.dev/control-protocol/pkg/reconciler
 knative.dev/control-protocol/pkg/service
-# knative.dev/eventing v0.23.1-0.20210608085741-b05d993d260e
+# knative.dev/eventing v0.23.1-0.20210608130141-92d85bf916e9
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test


Test patch:
diff --git a/pkg/source/mtadapter/adapter_test.go b/pkg/source/mtadapter/adapter_test.go
index 6250134cc..f29e6025f 100644
--- a/pkg/source/mtadapter/adapter_test.go
+++ b/pkg/source/mtadapter/adapter_test.go
@@ -18,17 +18,14 @@ 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"
 
@@ -54,7 +51,7 @@ func TestUpdateRemoveSources(t *testing.T) {
 	ctx, _ := pkgtesting.SetupFakeContext(t)
 	ctx, cancelAdapter := context.WithCancel(ctx)
 
-	env := &AdapterConfig{PodName: podName, MemoryRequest: "0"}
+	env := &AdapterConfig{PodName: podName, MemoryLimit: "0"}
 	ceClient := adaptertest.NewTestClient()
 
 	mtadapter := newAdapter(ctx, env, ceClient, newSampleAdapter).(*Adapter)
@@ -355,7 +352,7 @@ func TestSourceMTAdapter(t *testing.T) {
 
 			ctx, cancelAdapter := context.WithCancel(ctx)
 
-			env := &AdapterConfig{PodName: podName, MemoryRequest: "0"}
+			env := &AdapterConfig{PodName: podName, MemoryLimit: "0"}
 			ceClient := adaptertest.NewTestClient()
 
 			adapter := newAdapter(ctx, env, ceClient, newSampleAdapter).(*Adapter)
@@ -408,40 +405,3 @@ 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 / maxBrokers},
-		"memory request, response size more 64k":      {memoryRequest: 512 * 1024, sourceCount: 4, want: 128 * 1024 / maxBrokers},
-		"memory request, response size more than cap": {memoryRequest: 100 * 1024 * 1024, sourceCount: 1, want: responseSizeCap / maxBrokers},
-	}
-
-	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)
-			}
-		})
-	}
-
-}