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

Issue:
[scheduler] mtadapters may not scale down 
**Describe the bug**
The scheduler does not reassign vreplicas to "lower" pods  (ie: with a lower ordinal), preventing the autoscaler to scale down the mtadapter. 

 **Expected behavior**
A clear and concise description of what you expected to happen.
The mtadapter should be able to scale down when there are unused pods.

 **To Reproduce**
Steps to reproduce the behavior.
Create 1 KafkaSource CR and scale it to 150. In knative-eventing, observe 2 adapter replicas. Create a second KafkaSource CR. Scale down the first KafkaSource CR to 1. The total number of vreplicas is now 2, spread across the 2 adapter replicas. In theory only one adapter is needed. 

 **Knative release version**

 **Additional context**
Add any other context about the problem here such as proposed priority


Pull request description:
[scheduler] Compact vreplicas
Fixes #699

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- The statefulset autoscaler now attempts to compact vreplicas (MaxFillUp strategy only)
-
-

<!--
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
🎁 The multi-tenant KafkaSource controller periodically attempts to relocate KafkaSource consumers to Statefulset pods with a lower ordinal
```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


Patch:
diff --git a/go.mod b/go.mod
index c375f2e0a..94a6cef6e 100644
--- a/go.mod
+++ b/go.mod
@@ -35,9 +35,9 @@ require (
 	k8s.io/client-go v0.20.7
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920
 	knative.dev/control-protocol v0.0.0-20210616092621-1469a722be79
-	knative.dev/eventing v0.23.1-0.20210615213121-4eb1738b5e35
+	knative.dev/eventing v0.23.1-0.20210616164922-6adafe56e55e
 	knative.dev/hack v0.0.0-20210614141220-66ab1a098940
 	knative.dev/networking v0.0.0-20210615114921-e291c8011a20
-	knative.dev/pkg v0.0.0-20210615143321-77ff8d962c73
+	knative.dev/pkg v0.0.0-20210616195222-841aa7369ca1
 	knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7
 )
diff --git a/go.sum b/go.sum
index 95a0e43b0..f24bc2fe0 100644
--- a/go.sum
+++ b/go.sum
@@ -1348,8 +1348,8 @@ k8s.io/utils v0.0.0-20201110183641-67b214c5f920 h1:CbnUZsM497iRC5QMVkHwyl8s2tB3g
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20210616092621-1469a722be79 h1:CepZkHLRpqnQtdbYtXa7R/V5LbjxzkKIQdQxNOFyrhY=
 knative.dev/control-protocol v0.0.0-20210616092621-1469a722be79/go.mod h1:iRE6O4mY88gH6xWl6ZFhG6sJobzrZjFrz5xhX4h93Ns=
-knative.dev/eventing v0.23.1-0.20210615213121-4eb1738b5e35 h1:+XbEA0ge3iiWi7on4tLhM3T14YR7FtC8ts2GPAVvAYk=
-knative.dev/eventing v0.23.1-0.20210615213121-4eb1738b5e35/go.mod h1:d4zElgG7AL4zDt7CY7aW6BHdITgkUifk7Hskl5p1JEI=
+knative.dev/eventing v0.23.1-0.20210616164922-6adafe56e55e h1:+HmAva9pif4dbBXuRIXNXmjQLByemVHxg57SJzx5fPU=
+knative.dev/eventing v0.23.1-0.20210616164922-6adafe56e55e/go.mod h1:8fnNPdQgBT7NT+yCFgE1kuvN0mMsfnCfPIwnlyPrxy0=
 knative.dev/hack v0.0.0-20210601210329-de04b70e00d0/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20210614141220-66ab1a098940 h1:Aao7SsI087Pa8x+LDLaBdj/ylduPh2bJ8HbB7UjXElE=
 knative.dev/hack v0.0.0-20210614141220-66ab1a098940/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
@@ -1358,8 +1358,9 @@ knative.dev/networking v0.0.0-20210615114921-e291c8011a20 h1:TjlBAi9Nvoqh/Guvtye
 knative.dev/networking v0.0.0-20210615114921-e291c8011a20/go.mod h1:TH8itwXTBZ5L8mFOwPot/t9LrklYGwL5dkAl6COshj0=
 knative.dev/pkg v0.0.0-20210602095030-0e61d6763dd6/go.mod h1:/hT1IXHPs8ZwQ8yTJtfaYWpLmsLhC2s6/v/ydF7PoiE=
 knative.dev/pkg v0.0.0-20210615092720-192b0c9d6e56/go.mod h1:QVfbwA5ovshGnXDcx54l2QPYhGKppw/JgLYJWe7sA6A=
-knative.dev/pkg v0.0.0-20210615143321-77ff8d962c73 h1:p7KMCn1+n9QtJ/YZLncpMwSlWKv8vUA5IpiCaCxJWYY=
 knative.dev/pkg v0.0.0-20210615143321-77ff8d962c73/go.mod h1:QVfbwA5ovshGnXDcx54l2QPYhGKppw/JgLYJWe7sA6A=
+knative.dev/pkg v0.0.0-20210616195222-841aa7369ca1 h1:yLdxPtfiJ7UpfK99U4lgTe9L0wwWewqTELqMq4NhqHg=
+knative.dev/pkg v0.0.0-20210616195222-841aa7369ca1/go.mod h1:QVfbwA5ovshGnXDcx54l2QPYhGKppw/JgLYJWe7sA6A=
 knative.dev/reconciler-test v0.0.0-20210602211045-db062eb7ce42/go.mod h1:Ggh7/FyaVr9R1kQiuB9KpNsvaKVoPlZ0w8OoxUGi9Ys=
 knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7 h1:BNP1vs/FNDfpqFb+FaDzwZ9qnr3X+ms5u3lZDZk2ivw=
 knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7/go.mod h1:Ggh7/FyaVr9R1kQiuB9KpNsvaKVoPlZ0w8OoxUGi9Ys=
diff --git a/pkg/channel/consolidated/dispatcher/consumer_message_handler.go b/pkg/channel/consolidated/dispatcher/consumer_message_handler.go
index 80aef9d53..641db4fd2 100644
--- a/pkg/channel/consolidated/dispatcher/consumer_message_handler.go
+++ b/pkg/channel/consolidated/dispatcher/consumer_message_handler.go
@@ -23,7 +23,6 @@ import (
 	"github.com/Shopify/sarama"
 	protocolkafka "github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
-	"github.com/cloudevents/sdk-go/v2/binding/buffering"
 	"go.uber.org/zap"
 	"knative.dev/eventing-kafka/pkg/common/consumer"
 	"knative.dev/eventing-kafka/pkg/common/tracing"
@@ -76,29 +75,21 @@ func (c consumerMessageHandler) Handle(ctx context.Context, consumerMessage *sar
 
 	te := kncloudevents.TypeExtractorTransformer("")
 
-	bufferedMessage, err := buffering.CopyMessage(ctx, message, &te)
-
-	if err != nil {
-		return false, err
-	}
-
-	args := eventingchannels.ReportArgs{
-		Ns:        c.channelNs,
-		EventType: string(te),
-	}
-
-	_ = message.Finish(nil)
-
 	dispatchExecutionInfo, err := c.dispatcher.DispatchMessageWithRetries(
 		ctx,
-		bufferedMessage,
+		message,
 		nil,
 		c.sub.Subscriber,
 		c.sub.Reply,
 		c.sub.DeadLetter,
 		c.sub.RetryConfig,
+		&te,
 	)
 
+	args := eventingchannels.ReportArgs{
+		Ns:        c.channelNs,
+		EventType: string(te),
+	}
 	_ = fanout.ParseDispatchResultAndReportMetrics(fanout.NewDispatchResult(err, dispatchExecutionInfo), c.reporter, args)
 
 	// NOTE: only return `true` here if DispatchMessage actually delivered the message.
diff --git a/pkg/common/scheduler/scheduler.go b/pkg/common/scheduler/scheduler.go
index 3bd40cf2a..f4234f90d 100644
--- a/pkg/common/scheduler/scheduler.go
+++ b/pkg/common/scheduler/scheduler.go
@@ -31,6 +31,11 @@ var (
 // VPodLister is the function signature for returning a list of VPods
 type VPodLister func() ([]VPod, error)
 
+// Evictor allows for vreplicas to be evicted.
+// For instance, the evictor is used by the statefulset scheduler to
+// move vreplicas to pod with a lower ordinal.
+type Evictor func(vpod VPod, from *duckv1alpha1.Placement) error
+
 // Scheduler is responsible for placing VPods into real Kubernetes pods
 type Scheduler interface {
 	// Schedule computes the new set of placements for vpod.
diff --git a/pkg/common/scheduler/statefulset/autoscaler.go b/pkg/common/scheduler/statefulset/autoscaler.go
index 5923f35f0..6de54d327 100644
--- a/pkg/common/scheduler/statefulset/autoscaler.go
+++ b/pkg/common/scheduler/statefulset/autoscaler.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	clientappsv1 "k8s.io/client-go/kubernetes/typed/apps/v1"
 
+	"knative.dev/eventing-kafka/pkg/common/scheduler"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/logging"
 )
@@ -42,9 +43,11 @@ type Autoscaler interface {
 type autoscaler struct {
 	statefulSetClient clientappsv1.StatefulSetInterface
 	statefulSetName   string
+	vpodLister        scheduler.VPodLister
 	logger            *zap.SugaredLogger
 	stateAccessor     stateAccessor
 	trigger           chan int32
+	evictor           scheduler.Evictor
 
 	// capacity is the total number of virtual replicas available per pod.
 	capacity int32
@@ -55,7 +58,9 @@ type autoscaler struct {
 
 func NewAutoscaler(ctx context.Context,
 	namespace, name string,
+	lister scheduler.VPodLister,
 	stateAccessor stateAccessor,
+	evictor scheduler.Evictor,
 	refreshPeriod time.Duration,
 	capacity int32) Autoscaler {
 
@@ -63,7 +68,9 @@ func NewAutoscaler(ctx context.Context,
 		logger:            logging.FromContext(ctx),
 		statefulSetClient: kubeclient.Get(ctx).AppsV1().StatefulSets(namespace),
 		statefulSetName:   name,
+		vpodLister:        lister,
 		stateAccessor:     stateAccessor,
+		evictor:           evictor,
 		trigger:           make(chan int32, 1),
 		capacity:          capacity,
 		refreshPeriod:     refreshPeriod,
@@ -155,7 +162,64 @@ func (a *autoscaler) doautoscale(ctx context.Context, attemptScaleDown bool, pen
 			a.logger.Errorw("updating scale subresource failed", zap.Error(err))
 			return err
 		}
+	} else {
+		// since the number of replicas hasn't change, take the opportunity to
+		// compact the vreplicas
+		a.mayCompact(state)
 	}
 
 	return nil
 }
+
+func (a *autoscaler) mayCompact(s *state) {
+	// when there is only one pod there is nothing to move!
+	if s.lastOrdinal < 1 {
+		return
+	}
+
+	if s.schedulerPolicy == MAXFILLUP {
+		// Determine if there is enough free capacity to
+		// move all vreplicas placed in the last pod to pods with a lower ordinal
+		freeCapacity := s.freeCapacity() - s.Free(s.lastOrdinal)
+		usedInLastPod := s.capacity - s.Free(s.lastOrdinal)
+
+		if freeCapacity >= usedInLastPod {
+			err := a.compact(s)
+			if err != nil {
+				a.logger.Errorw("vreplicas compaction failed", zap.Error(err))
+			}
+		}
+
+		// only do 1 replica at a time to avoid overloading the scheduler with too many
+		// rescheduling requests.
+	}
+}
+
+func (a *autoscaler) compact(s *state) error {
+	a.logger.Info("compacting vreplicas")
+	vpods, err := a.vpodLister()
+	if err != nil {
+		return err
+	}
+
+	for _, vpod := range vpods {
+		placements := vpod.GetPlacements()
+		for _, placement := range placements {
+			ordinal := ordinalFromPodName(placement.PodName)
+
+			if ordinal == s.lastOrdinal {
+				a.logger.Infow("evicting vreplica(s)",
+					zap.String("name", vpod.GetKey().Name),
+					zap.String("namespace", vpod.GetKey().Namespace),
+					zap.String("podname", placement.PodName),
+					zap.Int("vreplicas", int(placement.VReplicas)))
+
+				err = a.evictor(vpod, &placement)
+				if err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}
diff --git a/pkg/common/scheduler/statefulset/scheduler.go b/pkg/common/scheduler/statefulset/scheduler.go
index 624c0d7f9..01c316b04 100644
--- a/pkg/common/scheduler/statefulset/scheduler.go
+++ b/pkg/common/scheduler/statefulset/scheduler.go
@@ -62,10 +62,11 @@ func NewScheduler(ctx context.Context,
 	refreshPeriod time.Duration,
 	capacity int32,
 	schedulerPolicy SchedulerPolicyType,
-	nodeLister corev1listers.NodeLister) scheduler.Scheduler {
+	nodeLister corev1listers.NodeLister,
+	evictor scheduler.Evictor) scheduler.Scheduler {
 
 	stateAccessor := newStateBuilder(ctx, lister, capacity, schedulerPolicy, nodeLister)
-	autoscaler := NewAutoscaler(ctx, namespace, name, stateAccessor, refreshPeriod, capacity)
+	autoscaler := NewAutoscaler(ctx, namespace, name, lister, stateAccessor, evictor, refreshPeriod, capacity)
 	podInformer := podinformer.Get(ctx)
 	podLister := podInformer.Lister().Pods(namespace)
 
@@ -136,7 +137,7 @@ func (s *StatefulSetScheduler) scheduleVPod(vpod scheduler.VPod) ([]duckv1alpha1
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
-	logger := s.logger.With("key", zap.Any("key", vpod.GetKey()))
+	logger := s.logger.With("key", vpod.GetKey())
 	logger.Info("scheduling")
 
 	// Get the current placements state
@@ -481,9 +482,8 @@ func (s *StatefulSetScheduler) updateStatefulset(obj interface{}) {
 
 	if statefulset.Spec.Replicas == nil {
 		s.replicas = 1
-	} else {
+	} else if s.replicas != *statefulset.Spec.Replicas {
 		s.replicas = *statefulset.Spec.Replicas
+		s.logger.Infow("statefulset replicas updated", zap.Int32("replicas", s.replicas))
 	}
-
-	s.logger.Infow("updated statefulset replicas", zap.Int32("replicas", s.replicas))
 }
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index 39f5e463a..a3f493cae 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -67,10 +67,9 @@ func NewEnvConfig() adapter.EnvConfigAccessor {
 }
 
 type Adapter struct {
-	config            *AdapterConfig
-	controlServer     *ctrlnetwork.ControlServer
-	saramaConfig      *sarama.Config
-	offsetInitialized bool
+	config        *AdapterConfig
+	controlServer *ctrlnetwork.ControlServer
+	saramaConfig  *sarama.Config
 
 	httpMessageSender *kncloudevents.HTTPMessageSender
 	reporter          pkgsource.StatsReporter
diff --git a/pkg/source/reconciler/mtsource/controller.go b/pkg/source/reconciler/mtsource/controller.go
index d17cde4c2..68f170565 100644
--- a/pkg/source/reconciler/mtsource/controller.go
+++ b/pkg/source/reconciler/mtsource/controller.go
@@ -18,9 +18,15 @@ package mtsource
 
 import (
 	"context"
+	"fmt"
 	"time"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"github.com/kelseyhightower/envconfig"
+	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/pkg/apis/duck"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
@@ -30,18 +36,20 @@ import (
 
 	"knative.dev/eventing/pkg/reconciler/source"
 
+	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	kafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client"
 	kafkainformer "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource"
 	"knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
-	scheduler "knative.dev/eventing-kafka/pkg/common/scheduler/statefulset"
+	scheduler "knative.dev/eventing-kafka/pkg/common/scheduler"
+	stsscheduler "knative.dev/eventing-kafka/pkg/common/scheduler/statefulset"
 	nodeinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/node"
 )
 
 type envConfig struct {
-	SchedulerRefreshPeriod int64                         `envconfig:"AUTOSCALER_REFRESH_PERIOD" required:"true"`
-	PodCapacity            int32                         `envconfig:"POD_CAPACITY" required:"true"`
-	SchedulerPolicy        scheduler.SchedulerPolicyType `envconfig:"SCHEDULER_POLICY_TYPE" required:"true"`
+	SchedulerRefreshPeriod int64                            `envconfig:"AUTOSCALER_REFRESH_PERIOD" required:"true"`
+	PodCapacity            int32                            `envconfig:"POD_CAPACITY" required:"true"`
+	SchedulerPolicy        stsscheduler.SchedulerPolicyType `envconfig:"SCHEDULER_POLICY_TYPE" required:"true"`
 }
 
 func NewController(
@@ -73,7 +81,53 @@ 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, env.SchedulerPolicy, nodeInformer.Lister())
+	evictor := func(vpod scheduler.VPod, from *duckv1alpha1.Placement) error {
+		key := vpod.GetKey()
+		sources := kafkaclient.Get(ctx).SourcesV1beta1().KafkaSources(key.Namespace)
+
+		before, err := sources.Get(ctx, key.Name, metav1.GetOptions{})
+		if err != nil {
+			return err
+		}
+
+		after := before.DeepCopy()
+
+		bp := after.GetPlacements()
+		ap := make([]duckv1alpha1.Placement, 0, len(bp)-1)
+		for _, p := range bp {
+			if p.PodName != from.PodName {
+				ap = append(ap, p)
+			}
+		}
+		after.Status.Placement = ap
+
+		jsonPatch, err := duck.CreatePatch(before, after)
+		if err != nil {
+			return err
+		}
+
+		// If there is nothing to patch, we are good, just return.
+		// Empty patch is [], hence we check for that.
+		if len(jsonPatch) == 0 {
+			return nil
+		}
+
+		patch, err := jsonPatch.MarshalJSON()
+		if err != nil {
+			return fmt.Errorf("marshaling JSON patch: %w", err)
+		}
+
+		patched, err := kafkaclient.Get(ctx).SourcesV1beta1().KafkaSources(key.Namespace).Patch(ctx, key.Name, types.JSONPatchType, patch, metav1.PatchOptions{}, "status")
+		if err != nil {
+			return fmt.Errorf("Failed patching: %w", err)
+		}
+		logging.FromContext(ctx).Debugw("Patched resource", zap.Any("patch", patch), zap.Any("patched", patched))
+		return nil
+	}
+
+	c.scheduler = stsscheduler.NewScheduler(ctx,
+		system.Namespace(), mtadapterName, c.vpodLister, rp, env.PodCapacity, env.SchedulerPolicy,
+		nodeInformer.Lister(), evictor)
 
 	logging.FromContext(ctx).Info("Setting up kafka event handlers")
 	kafkaInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
diff --git a/third_party/VENDOR-LICENSE/k8s.io/apiserver/pkg/storage/names/LICENSE b/third_party/VENDOR-LICENSE/k8s.io/apiserver/pkg/storage/names/LICENSE
deleted file mode 100644
index d64569567..000000000
--- a/third_party/VENDOR-LICENSE/k8s.io/apiserver/pkg/storage/names/LICENSE
+++ /dev/null
@@ -1,202 +0,0 @@
-
-                                 Apache License
-                           Version 2.0, January 2004
-                        http://www.apache.org/licenses/
-
-   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
-
-   1. Definitions.
-
-      "License" shall mean the terms and conditions for use, reproduction,
-      and distribution as defined by Sections 1 through 9 of this document.
-
-      "Licensor" shall mean the copyright owner or entity authorized by
-      the copyright owner that is granting the License.
-
-      "Legal Entity" shall mean the union of the acting entity and all
-      other entities that control, are controlled by, or are under common
-      control with that entity. For the purposes of this definition,
-      "control" means (i) the power, direct or indirect, to cause the
-      direction or management of such entity, whether by contract or
-      otherwise, or (ii) ownership of fifty percent (50%) or more of the
-      outstanding shares, or (iii) beneficial ownership of such entity.
-
-      "You" (or "Your") shall mean an individual or Legal Entity
-      exercising permissions granted by this License.
-
-      "Source" form shall mean the preferred form for making modifications,
-      including but not limited to software source code, documentation
-      source, and configuration files.
-
-      "Object" form shall mean any form resulting from mechanical
-      transformation or translation of a Source form, including but
-      not limited to compiled object code, generated documentation,
-      and conversions to other media types.
-
-      "Work" shall mean the work of authorship, whether in Source or
-      Object form, made available under the License, as indicated by a
-      copyright notice that is included in or attached to the work
-      (an example is provided in the Appendix below).
-
-      "Derivative Works" shall mean any work, whether in Source or Object
-      form, that is based on (or derived from) the Work and for which the
-      editorial revisions, annotations, elaborations, or other modifications
-      represent, as a whole, an original work of authorship. For the purposes
-      of this License, Derivative Works shall not include works that remain
-      separable from, or merely link (or bind by name) to the interfaces of,
-      the Work and Derivative Works thereof.
-
-      "Contribution" shall mean any work of authorship, including
-      the original version of the Work and any modifications or additions
-      to that Work or Derivative Works thereof, that is intentionally
-      submitted to Licensor for inclusion in the Work by the copyright owner
-      or by an individual or Legal Entity authorized to submit on behalf of
-      the copyright owner. For the purposes of this definition, "submitted"
-      means any form of electronic, verbal, or written communication sent
-      to the Licensor or its representatives, including but not limited to
-      communication on electronic mailing lists, source code control systems,
-      and issue tracking systems that are managed by, or on behalf of, the
-      Licensor for the purpose of discussing and improving the Work, but
-      excluding communication that is conspicuously marked or otherwise
-      designated in writing by the copyright owner as "Not a Contribution."
-
-      "Contributor" shall mean Licensor and any individual or Legal Entity
-      on behalf of whom a Contribution has been received by Licensor and
-      subsequently incorporated within the Work.
-
-   2. Grant of Copyright License. Subject to the terms and conditions of
-      this License, each Contributor hereby grants to You a perpetual,
-      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
-      copyright license to reproduce, prepare Derivative Works of,
-      publicly display, publicly perform, sublicense, and distribute the
-      Work and such Derivative Works in Source or Object form.
-
-   3. Grant of Patent License. Subject to the terms and conditions of
-      this License, each Contributor hereby grants to You a perpetual,
-      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
-      (except as stated in this section) patent license to make, have made,
-      use, offer to sell, sell, import, and otherwise transfer the Work,
-      where such license applies only to those patent claims licensable
-      by such Contributor that are necessarily infringed by their
-      Contribution(s) alone or by combination of their Contribution(s)
-      with the Work to which such Contribution(s) was submitted. If You
-      institute patent litigation against any entity (including a
-      cross-claim or counterclaim in a lawsuit) alleging that the Work
-      or a Contribution incorporated within the Work constitutes direct
-      or contributory patent infringement, then any patent licenses
-      granted to You under this License for that Work shall terminate
-      as of the date such litigation is filed.
-
-   4. Redistribution. You may reproduce and distribute copies of the
-      Work or Derivative Works thereof in any medium, with or without
-      modifications, and in Source or Object form, provided that You
-      meet the following conditions:
-
-      (a) You must give any other recipients of the Work or
-          Derivative Works a copy of this License; and
-
-      (b) You must cause any modified files to carry prominent notices
-          stating that You changed the files; and
-
-      (c) You must retain, in the Source form of any Derivative Works
-          that You distribute, all copyright, patent, trademark, and
-          attribution notices from the Source form of the Work,
-          excluding those notices that do not pertain to any part of
-          the Derivative Works; and
-
-      (d) If the Work includes a "NOTICE" text file as part of its
-          distribution, then any Derivative Works that You distribute must
-          include a readable copy of the attribution notices contained
-          within such NOTICE file, excluding those notices that do not
-          pertain to any part of the Derivative Works, in at least one
-          of the following places: within a NOTICE text file distributed
-          as part of the Derivative Works; within the Source form or
-          documentation, if provided along with the Derivative Works; or,
-          within a display generated by the Derivative Works, if and
-          wherever such third-party notices normally appear. The contents
-          of the NOTICE file are for informational purposes only and
-          do not modify the License. You may add Your own attribution
-          notices within Derivative Works that You distribute, alongside
-          or as an addendum to the NOTICE text from the Work, provided
-          that such additional attribution notices cannot be construed
-          as modifying the License.
-
-      You may add Your own copyright statement to Your modifications and
-      may provide additional or different license terms and conditions
-      for use, reproduction, or distribution of Your modifications, or
-      for any such Derivative Works as a whole, provided Your use,
-      reproduction, and distribution of the Work otherwise complies with
-      the conditions stated in this License.
-
-   5. Submission of Contributions. Unless You explicitly state otherwise,
-      any Contribution intentionally submitted for inclusion in the Work
-      by You to the Licensor shall be under the terms and conditions of
-      this License, without any additional terms or conditions.
-      Notwithstanding the above, nothing herein shall supersede or modify
-      the terms of any separate license agreement you may have executed
-      with Licensor regarding such Contributions.
-
-   6. Trademarks. This License does not grant permission to use the trade
-      names, trademarks, service marks, or product names of the Licensor,
-      except as required for reasonable and customary use in describing the
-      origin of the Work and reproducing the content of the NOTICE file.
-
-   7. Disclaimer of Warranty. Unless required by applicable law or
-      agreed to in writing, Licensor provides the Work (and each
-      Contributor provides its Contributions) on an "AS IS" BASIS,
-      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
-      implied, including, without limitation, any warranties or conditions
-      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
-      PARTICULAR PURPOSE. You are solely responsible for determining the
-      appropriateness of using or redistributing the Work and assume any
-      risks associated with Your exercise of permissions under this License.
-
-   8. Limitation of Liability. In no event and under no legal theory,
-      whether in tort (including negligence), contract, or otherwise,
-      unless required by applicable law (such as deliberate and grossly
-      negligent acts) or agreed to in writing, shall any Contributor be
-      liable to You for damages, including any direct, indirect, special,
-      incidental, or consequential damages of any character arising as a
-      result of this License or out of the use or inability to use the
-      Work (including but not limited to damages for loss of goodwill,
-      work stoppage, computer failure or malfunction, or any and all
-      other commercial damages or losses), even if such Contributor
-      has been advised of the possibility of such damages.
-
-   9. Accepting Warranty or Additional Liability. While redistributing
-      the Work or Derivative Works thereof, You may choose to offer,
-      and charge a fee for, acceptance of support, warranty, indemnity,
-      or other liability obligations and/or rights consistent with this
-      License. However, in accepting such obligations, You may act only
-      on Your own behalf and on Your sole responsibility, not on behalf
-      of any other Contributor, and only if You agree to indemnify,
-      defend, and hold each Contributor harmless for any liability
-      incurred by, or claims asserted against, such Contributor by reason
-      of your accepting any such warranty or additional liability.
-
-   END OF TERMS AND CONDITIONS
-
-   APPENDIX: How to apply the Apache License to your work.
-
-      To apply the Apache License to your work, attach the following
-      boilerplate notice, with the fields enclosed by brackets "[]"
-      replaced with your own identifying information. (Don't include
-      the brackets!)  The text should be enclosed in the appropriate
-      comment syntax for the file format. We also recommend that a
-      file or class name and description of purpose be included on the
-      same "printed page" as the copyright notice for easier
-      identification within third-party archives.
-
-   Copyright [yyyy] [name of copyright owner]
-
-   Licensed under the Apache License, Version 2.0 (the "License");
-   you may not use this file except in compliance with the License.
-   You may obtain a copy of the License at
-
-       http://www.apache.org/licenses/LICENSE-2.0
-
-   Unless required by applicable law or agreed to in writing, software
-   distributed under the License is distributed on an "AS IS" BASIS,
-   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-   See the License for the specific language governing permissions and
-   limitations under the License.
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 6977ad0a5..f8be65e62 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1118,7 +1118,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.20210615213121-4eb1738b5e35
+# knative.dev/eventing v0.23.1-0.20210616164922-6adafe56e55e
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test
@@ -1243,7 +1243,7 @@ knative.dev/hack/shell
 # knative.dev/networking v0.0.0-20210615114921-e291c8011a20
 ## explicit
 knative.dev/networking/pkg/prober
-# knative.dev/pkg v0.0.0-20210615143321-77ff8d962c73
+# knative.dev/pkg v0.0.0-20210616195222-841aa7369ca1
 ## explicit
 knative.dev/pkg/apiextensions/storageversion
 knative.dev/pkg/apiextensions/storageversion/cmd/migrate


Test patch:
diff --git a/pkg/common/scheduler/statefulset/autoscaler_test.go b/pkg/common/scheduler/statefulset/autoscaler_test.go
index 1c671eda1..924e8f840 100644
--- a/pkg/common/scheduler/statefulset/autoscaler_test.go
+++ b/pkg/common/scheduler/statefulset/autoscaler_test.go
@@ -17,11 +17,13 @@ limitations under the License.
 package statefulset
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	gtesting "k8s.io/client-go/testing"
 
 	listers "knative.dev/eventing/pkg/reconciler/testing/v1"
@@ -222,7 +224,11 @@ func TestAutoscaler(t *testing.T) {
 				t.Fatal("unexpected error", err)
 			}
 
-			autoscaler := NewAutoscaler(ctx, testNs, sfsName, stateAccessor, 10*time.Second, int32(10)).(*autoscaler)
+			noopEvictor := func(vpod scheduler.VPod, from *duckv1alpha1.Placement) error {
+				return nil
+			}
+
+			autoscaler := NewAutoscaler(ctx, testNs, sfsName, vpodClient.List, stateAccessor, noopEvictor, 10*time.Second, int32(10)).(*autoscaler)
 
 			for _, vpod := range tc.vpods {
 				vpodClient.Append(vpod)
@@ -266,7 +272,11 @@ func TestAutoscalerScaleDownToZero(t *testing.T) {
 		t.Fatal("unexpected error", err)
 	}
 
-	autoscaler := NewAutoscaler(ctx, testNs, sfsName, stateAccessor, 2*time.Second, int32(10)).(*autoscaler)
+	noopEvictor := func(vpod scheduler.VPod, from *duckv1alpha1.Placement) error {
+		return nil
+	}
+
+	autoscaler := NewAutoscaler(ctx, testNs, sfsName, vpodClient.List, stateAccessor, noopEvictor, 2*time.Second, int32(10)).(*autoscaler)
 
 	done := make(chan bool)
 	go func() {
@@ -297,3 +307,141 @@ func TestAutoscalerScaleDownToZero(t *testing.T) {
 		t.Fatal("timeout waiting for autoscaler to stop")
 	}
 }
+
+func TestCompactor(t *testing.T) {
+	testCases := []struct {
+		name            string
+		replicas        int32
+		vpods           []scheduler.VPod
+		schedulerPolicy SchedulerPolicyType
+		wantEvictions   map[types.NamespacedName]duckv1alpha1.Placement
+	}{
+		{
+			name:     "no replicas, no placements, no pending",
+			replicas: int32(0),
+			vpods: []scheduler.VPod{
+				tscheduler.NewVPod(testNs, "vpod-1", 0, nil),
+			},
+			schedulerPolicy: MAXFILLUP,
+			wantEvictions:   nil,
+		},
+		{
+			name:     "one vpod, with placements in 2 pods, compacted",
+			replicas: int32(2),
+			vpods: []scheduler.VPod{
+				tscheduler.NewVPod(testNs, "vpod-1", 15, []duckv1alpha1.Placement{
+					{PodName: "pod-0", VReplicas: int32(8)},
+					{PodName: "pod-1", VReplicas: int32(7)}}),
+			},
+			schedulerPolicy: MAXFILLUP,
+			wantEvictions:   nil,
+		},
+		{
+			name:     "one vpod, with  placements in 2 pods, compacted edge",
+			replicas: int32(2),
+			vpods: []scheduler.VPod{
+				tscheduler.NewVPod(testNs, "vpod-1", 11, []duckv1alpha1.Placement{
+					{PodName: "pod-0", VReplicas: int32(8)},
+					{PodName: "pod-1", VReplicas: int32(3)}}),
+			},
+			schedulerPolicy: MAXFILLUP,
+			wantEvictions:   nil,
+		},
+		{
+			name:     "one vpod, with placements in 2 pods, not compacted",
+			replicas: int32(2),
+			vpods: []scheduler.VPod{
+				tscheduler.NewVPod(testNs, "vpod-1", 10, []duckv1alpha1.Placement{
+					{PodName: "pod-0", VReplicas: int32(8)},
+					{PodName: "pod-1", VReplicas: int32(2)}}),
+			},
+			schedulerPolicy: MAXFILLUP,
+			wantEvictions: map[types.NamespacedName]duckv1alpha1.Placement{
+				{Name: "vpod-1", Namespace: testNs}: {PodName: "pod-1", VReplicas: int32(2)},
+			},
+		},
+		{
+			name:     "multiple vpods, with placements in multiple pods, compacted",
+			replicas: int32(2),
+			// pod-0:6, pod-1:8, pod-2:7
+			vpods: []scheduler.VPod{
+				tscheduler.NewVPod(testNs, "vpod-1", 12, []duckv1alpha1.Placement{
+					{PodName: "pod-0", VReplicas: int32(4)},
+					{PodName: "pod-1", VReplicas: int32(8)}}),
+				tscheduler.NewVPod(testNs, "vpod-2", 9, []duckv1alpha1.Placement{
+					{PodName: "pod-0", VReplicas: int32(2)},
+					{PodName: "pod-2", VReplicas: int32(7)}}),
+			},
+			schedulerPolicy: MAXFILLUP,
+			wantEvictions:   nil,
+		},
+		{
+			name:     "multiple vpods, with placements in multiple pods, not compacted",
+			replicas: int32(2),
+			// pod-0:6, pod-1:7, pod-2:7
+			vpods: []scheduler.VPod{
+				tscheduler.NewVPod(testNs, "vpod-1", 6, []duckv1alpha1.Placement{
+					{PodName: "pod-0", VReplicas: int32(4)},
+					{PodName: "pod-1", VReplicas: int32(7)}}),
+				tscheduler.NewVPod(testNs, "vpod-2", 15, []duckv1alpha1.Placement{
+					{PodName: "pod-0", VReplicas: int32(2)},
+					{PodName: "pod-2", VReplicas: int32(7)}}),
+			},
+			schedulerPolicy: MAXFILLUP,
+			wantEvictions: map[types.NamespacedName]duckv1alpha1.Placement{
+				{Name: "vpod-2", Namespace: testNs}: {PodName: "pod-2", VReplicas: int32(7)},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx, _ := setupFakeContext(t)
+
+			vpodClient := tscheduler.NewVPodClient()
+			ls := listers.NewListers(nil)
+			stateAccessor := newStateBuilder(ctx, vpodClient.List, 10, tc.schedulerPolicy, ls.GetNodeLister())
+
+			evictions := make(map[types.NamespacedName]duckv1alpha1.Placement)
+			recordEviction := func(vpod scheduler.VPod, from *duckv1alpha1.Placement) error {
+				evictions[vpod.GetKey()] = *from
+				return nil
+			}
+
+			autoscaler := NewAutoscaler(ctx, testNs, sfsName, vpodClient.List, stateAccessor, recordEviction, 10*time.Second, int32(10)).(*autoscaler)
+
+			for _, vpod := range tc.vpods {
+				vpodClient.Append(vpod)
+			}
+
+			state, err := stateAccessor.State()
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			fmt.Println(state)
+
+			autoscaler.mayCompact(state)
+
+			if tc.wantEvictions == nil && len(evictions) != 0 {
+				t.Fatalf("unexpected evictions: %v", evictions)
+
+			}
+			for key, placement := range tc.wantEvictions {
+				got, ok := evictions[key]
+				if !ok {
+					t.Fatalf("unexpected %v to be evicted but was not", key)
+				}
+
+				if got != placement {
+					t.Fatalf("expected evicted placement to be %v, but got %v", placement, got)
+				}
+
+				delete(evictions, key)
+			}
+
+			if len(evictions) != 0 {
+				t.Fatalf("unexpected evictions %v", evictions)
+			}
+		})
+	}
+}
diff --git a/pkg/source/adapter/adapter_test.go b/pkg/source/adapter/adapter_test.go
index 1a9baad78..fdaf3843a 100644
--- a/pkg/source/adapter/adapter_test.go
+++ b/pkg/source/adapter/adapter_test.go
@@ -442,64 +442,3 @@ func TestAdapter_Start(t *testing.T) { // just increase code coverage
 	}
 	cancel()
 }
-
-type sampleConsumerSession struct {
-	offsets   map[string]map[int32]int64
-	committed bool
-}
-
-func (s *sampleConsumerSession) Claims() map[string][]int32 {
-	return nil
-}
-
-func (s *sampleConsumerSession) MemberID() string {
-	return ""
-}
-
-func (s *sampleConsumerSession) GenerationID() int32 {
-	return int32(0)
-}
-
-func (s *sampleConsumerSession) GetOffset(topic string, partition int32) int64 {
-	if s.offsets == nil {
-		return -2
-	}
-	partitions, ok := s.offsets[topic]
-	if !ok {
-		return -2
-	}
-	offset, ok := partitions[partition]
-	if !ok {
-		return -2
-	}
-	return offset
-}
-
-func (s *sampleConsumerSession) MarkOffset(topic string, partition int32, offset int64, metadata string) {
-	if s.offsets == nil {
-		s.offsets = make(map[string]map[int32]int64)
-	}
-
-	partitions, ok := s.offsets[topic]
-	if !ok {
-		partitions = make(map[int32]int64)
-		s.offsets[topic] = partitions
-	}
-	partitions[partition] = offset
-}
-
-func (s *sampleConsumerSession) Commit() {
-	s.committed = true
-}
-
-func (s *sampleConsumerSession) ResetOffset(topic string, partition int32, offset int64, metadata string) {
-
-}
-
-func (s *sampleConsumerSession) MarkMessage(msg *sarama.ConsumerMessage, metadata string) {
-
-}
-
-func (s *sampleConsumerSession) Context() context.Context {
-	return nil
-}
diff --git a/test/rekt/features/kafkasource/mt_data_plane.go b/test/rekt/features/kafkasource/mt_data_plane.go
new file mode 100644
index 000000000..bdee8ac45
--- /dev/null
+++ b/test/rekt/features/kafkasource/mt_data_plane.go
@@ -0,0 +1,144 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kafkasource
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/cloudevents/sdk-go/v2/event"
+	. "github.com/cloudevents/sdk-go/v2/test"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/eventshub/assert"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/state"
+
+	kafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client"
+	"knative.dev/eventing-kafka/test/rekt/resources/kafkacat"
+	"knative.dev/eventing-kafka/test/rekt/resources/kafkasource"
+	"knative.dev/eventing-kafka/test/rekt/resources/kafkatopic"
+)
+
+var test_mt_source = os.Getenv("TEST_MT_SOURCE")
+
+func MtDataPlaneDelivery() *feature.Feature {
+	name := "mt-tenant-test-plain"
+	f := feature.NewFeatureNamed("Delivery/" + name)
+	delivery := delivery{
+		message: kafkaMessage{
+			Key:     "0",
+			Payload: `{"value":711}`,
+		},
+		matchers: AllOf(
+			HasData([]byte(`{"value":711}`)),
+			HasExtension("key", "0"),
+		),
+	}
+
+	delivery.bootstrapServer = auths["plain"].bootstrapServer
+	delivery.auth = auths["plain"]
+
+	f.Setup("set kafka message", func(ctx context.Context, t feature.T) {
+		state.SetOrFail(ctx, t, "message", delivery.message)
+	})
+	f.Setup("set bootstrap server", func(ctx context.Context, t feature.T) {
+		state.SetOrFail(ctx, t, "bootstrapServer", delivery.bootstrapServer)
+	})
+
+	topicName := feature.MakeRandomK8sName("topic") // A k8s name is also a valid topic name.
+
+	f.Setup("set kafka topic name", func(ctx context.Context, t feature.T) {
+		state.SetOrFail(ctx, t, "topicName", topicName)
+	})
+
+	f.Setup("install kafka topic", kafkatopic.Install(topicName))
+
+	sinkName := feature.MakeRandomK8sName("sink")
+	f.Setup("set sink name", func(ctx context.Context, t feature.T) {
+		state.SetOrFail(ctx, t, "sinkName", sinkName)
+	})
+
+	f.Setup("install sink", eventshub.Install(sinkName, eventshub.StartReceiver))
+
+	ksopts := []kafkasource.CfgFn{
+		kafkasource.WithBootstrapServers([]string{delivery.bootstrapServer}),
+		kafkasource.WithTopics([]string{topicName}),
+		kafkasource.WithSink(&duckv1.KReference{
+			Kind:       "Service",
+			Name:       sinkName,
+			APIVersion: "v1",
+		}, ""),
+	}
+
+	f.Setup("install a kafka source", kafkasource.Install(name, ksopts...))
+	f.Requirement("kafka topic must be ready", kafkatopic.IsReady(topicName))
+	f.Requirement("kafka source must be ready", kafkasource.IsReady(name))
+	f.Assert("different KafkaSource exists on the same pod", sourcesPlaced(name, delivery.matchers))
+
+	return f
+}
+
+func sourcesPlaced(name string, matchers EventMatcher) func(ctx context.Context, t feature.T) {
+	return func(ctx context.Context, t feature.T) {
+		var message kafkaMessage
+		state.GetOrFail(ctx, t, "message", &message)
+
+		topicName := state.GetStringOrFail(ctx, t, "topicName")
+		sinkName := state.GetStringOrFail(ctx, t, "sinkName")
+
+		if test_mt_source == "1" {
+			time.Sleep(10 * time.Second)
+		}
+
+		kcopts := []kafkacat.CfgFn{
+			kafkacat.WithBootstrapServer(kafkaBootstrapUrlPlain),
+			kafkacat.WithTopic(topicName),
+			kafkacat.WithKey(message.Key),
+			kafkacat.WithHeaders(message.Headers),
+			kafkacat.WithPayload(message.Payload),
+		}
+		env := environment.FromContext(ctx)
+		matchers = AllOf(
+			func(have event.Event) error {
+				sourceInterface := kafkaclient.Get(ctx).SourcesV1beta1()
+
+				source1, err := sourceInterface.KafkaSources(env.Namespace()).Get(ctx, "mt-tenant-test-plain", metav1.GetOptions{})
+				if err != nil {
+					return fmt.Errorf("Error when getting Kafka source1")
+				}
+
+				source2, err := sourceInterface.KafkaSources(env.Namespace()).Get(ctx, "no-ce-event-plain", metav1.GetOptions{})
+				if err != nil {
+					return fmt.Errorf("Error when getting Kafka source2")
+				}
+
+				if source1.Status.Placement[0].PodName != source2.Status.Placement[0].PodName && len(source2.Status.Placement[0].PodName) != 0 {
+					return fmt.Errorf("Two Kafka source do not belong to the same pod")
+				}
+
+				return nil
+			})
+
+		kafkacat.Install(topicName, kcopts...)(ctx, t)
+		assert.OnStore(sinkName).MatchEvent(matchers).Exact(1)(ctx, t)
+	}
+}
diff --git a/test/rekt/kafka_source_test.go b/test/rekt/kafka_source_test.go
index 659043c07..cb41647fc 100644
--- a/test/rekt/kafka_source_test.go
+++ b/test/rekt/kafka_source_test.go
@@ -19,6 +19,7 @@ limitations under the License.
 package rekt
 
 import (
+	"os"
 	"testing"
 	"time"
 
@@ -36,9 +37,10 @@ const (
 	kafkaTLSSecret  = "strimzi-tls-secret"
 )
 
+var test_mt_soure = os.Getenv("TEST_MT_SOURCE")
+
 func TestKafkaSource(t *testing.T) {
 	t.Parallel()
-
 	ctx, env := global.Environment(
 		knative.WithKnativeNamespace(system.Namespace()),
 		knative.WithLoggingConfig,
@@ -59,7 +61,12 @@ func TestKafkaSource(t *testing.T) {
 		t.Fatalf("could not copy secret(%s): %v", kafkaSASLSecret, err)
 	}
 
-	for _, f := range kafkasource.DataPlaneDelivery() {
+	testFeatures := kafkasource.DataPlaneDelivery()
+	if test_mt_soure == "1" {
+		testFeatures = append(testFeatures, kafkasource.MtDataPlaneDelivery())
+	}
+
+	for _, f := range testFeatures {
 		env.Test(ctx, t, f)
 	}
 }
diff --git a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/factory.go b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/factory.go
index d384bfd62..753d0644d 100644
--- a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/factory.go
+++ b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/factory.go
@@ -58,7 +58,12 @@ func MakeFactory(ctor Ctor, unstructured bool, logger *zap.SugaredLogger) Factor
 	return func(t *testing.T, r *TableRow) (controller.Reconciler, ActionRecorderList, EventList) {
 		ls := NewListers(r.Objects)
 
-		ctx := context.Background()
+		var ctx context.Context
+		if r.Ctx != nil {
+			ctx = r.Ctx
+		} else {
+			ctx = context.Background()
+		}
 		ctx = logging.WithLogger(ctx, logger)
 
 		ctx, kubeClient := fakekubeclient.With(ctx, ls.GetKubeObjects()...)
diff --git a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go
index b4dd4083a..2d665b254 100644
--- a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go
+++ b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go
@@ -138,6 +138,19 @@ func WithSubscriptionSubscriberRef(gvk metav1.GroupVersionKind, name, namespace
 	}
 }
 
+func WithSubscriptionSubscriberRefUsingGroup(gvk metav1.GroupVersionKind, name, namespace string) SubscriptionOption {
+	return func(s *v1.Subscription) {
+		s.Spec.Subscriber = &duckv1.Destination{
+			Ref: &duckv1.KReference{
+				Group:     gvk.Group,
+				Kind:      gvk.Kind,
+				Name:      name,
+				Namespace: namespace,
+			},
+		}
+	}
+}
+
 func WithSubscriptionDeliveryRef(gvk metav1.GroupVersionKind, name, namespace string) SubscriptionOption {
 	return func(s *v1.Subscription) {
 		s.Spec.Delivery = &eventingduckv1.DeliverySpec{
diff --git a/vendor/knative.dev/eventing/test/e2e_flags.go b/vendor/knative.dev/eventing/test/e2e_flags.go
index 0abd69d85..b6dc34532 100644
--- a/vendor/knative.dev/eventing/test/e2e_flags.go
+++ b/vendor/knative.dev/eventing/test/e2e_flags.go
@@ -60,6 +60,9 @@ func InitializeEventingFlags() {
 	flag.Var(&EventingFlags.Brokers, "brokers", BrokerUsage)
 	flag.StringVar(&EventingFlags.BrokerName, "brokername", "", BrokerNameUsage)
 	flag.StringVar(&EventingFlags.BrokerNamespace, "brokernamespace", "", BrokerNamespaceUsage)
+	// Might be useful in restricted environments where namespaces need to be
+	// created by a user with increased privileges (admin).
+	flag.BoolVar(&EventingFlags.ReuseNamespace, "reusenamespace", false, "Whether to re-use namespace for a test if it already exists.")
 	flag.Parse()
 
 	// If no channel is passed through the flag, initialize it as the DefaultChannel.
diff --git a/vendor/knative.dev/eventing/test/flags/eventing_environment.go b/vendor/knative.dev/eventing/test/flags/eventing_environment.go
index 322060d53..83e5efd84 100644
--- a/vendor/knative.dev/eventing/test/flags/eventing_environment.go
+++ b/vendor/knative.dev/eventing/test/flags/eventing_environment.go
@@ -26,4 +26,5 @@ type EventingEnvironmentFlags struct {
 	ReadyFile       string
 	BrokerName      string
 	BrokerNamespace string
+	ReuseNamespace  bool
 }
diff --git a/vendor/knative.dev/eventing/test/lib/test_runner.go b/vendor/knative.dev/eventing/test/lib/test_runner.go
index 4066da298..986e58598 100644
--- a/vendor/knative.dev/eventing/test/lib/test_runner.go
+++ b/vendor/knative.dev/eventing/test/lib/test_runner.go
@@ -18,10 +18,13 @@ package lib
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"os"
 	"path/filepath"
 	"sort"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -29,10 +32,7 @@ import (
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/apiserver/pkg/storage/names"
-
 	pkgTest "knative.dev/pkg/test"
-	"knative.dev/pkg/test/helpers"
 	"knative.dev/pkg/test/prow"
 
 	"knative.dev/eventing/pkg/utils"
@@ -47,6 +47,15 @@ import (
 const (
 	podLogsDir         = "pod-logs"
 	testPullSecretName = "kn-eventing-test-pull-secret"
+	MaxNamespaceSkip   = 20
+	MaxRetries         = 5
+	RetrySleepDuration = 2 * time.Second
+)
+
+var (
+	nsMutex        sync.Mutex
+	namespaceCount int
+	ReuseNamespace bool
 )
 
 // ComponentsTestRunner is used to run tests against different eventing components.
@@ -148,19 +157,16 @@ var SetupClientOptionNoop SetupClientOption = func(*Client) {
 // Setup creates the client objects needed in the e2e tests,
 // and does other setups, like creating namespaces, set the test case to run in parallel, etc.
 func Setup(t *testing.T, runInParallel bool, options ...SetupClientOption) *Client {
-	// Create a new namespace to run this test case.
-	namespace := makeK8sNamespace(t.Name())
-	t.Logf("namespace is : %q", namespace)
-	client, err := NewClient(
-		pkgTest.Flags.Kubeconfig,
-		pkgTest.Flags.Cluster,
-		namespace,
-		t)
+	client, err := CreateNamespacedClient(t)
 	if err != nil {
 		t.Fatal("Couldn't initialize clients:", err)
 	}
 
-	CreateNamespaceIfNeeded(t, client, namespace)
+	// If namespaces are re-used the pull-secret is supposed to be created in advance.
+	if !ReuseNamespace {
+		SetupServiceAccount(t, client)
+		SetupPullSecret(t, client)
+	}
 
 	// Run the test case in parallel if needed.
 	if runInParallel {
@@ -178,9 +184,73 @@ func Setup(t *testing.T, runInParallel bool, options ...SetupClientOption) *Clie
 	return client
 }
 
-func makeK8sNamespace(baseFuncName string) string {
-	base := helpers.MakeK8sNamePrefix(baseFuncName)
-	return names.SimpleNameGenerator.GenerateName(base + "-")
+func CreateNamespacedClient(t *testing.T) (*Client, error) {
+	ns := ""
+	// Try next MaxNamespaceSkip namespaces before giving up. This should address the issue with
+	// development cycles when namespaces from previous runs were not cleaned properly.
+	for i := 0; i < MaxNamespaceSkip; i++ {
+		ns = NextNamespace()
+		client, err := NewClient(
+			pkgTest.Flags.Kubeconfig,
+			pkgTest.Flags.Cluster,
+			ns,
+			t)
+		if err != nil {
+			return nil, err
+		}
+		if ReuseNamespace {
+			// Re-using existing namespace, no need to create it.
+			// The namespace is supposed to be created in advance.
+			return client, nil
+		} else {
+			// The test is supposed to create a new test namespace for itself.
+			// Keep trying until we find a namespace that doesn't exist yet.
+			if err := CreateNamespaceWithRetry(client, ns); err != nil {
+				if apierrs.IsAlreadyExists(err) {
+					continue
+				}
+				return nil, err
+			}
+		}
+		return client, nil
+	}
+	return nil, errors.New("unable to find available namespace")
+}
+
+// NextNamespace returns the next unique namespace.
+func NextNamespace() string {
+	ns := os.Getenv("EVENTING_E2E_NAMESPACE")
+	if ns == "" {
+		ns = "eventing-e2e"
+	}
+	return fmt.Sprintf("%s%d", ns, GetNextNamespaceId())
+}
+
+// GetNextNamespaceId return the next unique ID for the next namespace.
+func GetNextNamespaceId() int {
+	nsMutex.Lock()
+	defer nsMutex.Unlock()
+	current := namespaceCount
+	namespaceCount++
+	return current
+}
+
+// CreateNamespaceWithRetry creates the given namespace with retries.
+func CreateNamespaceWithRetry(client *Client, namespace string) error {
+	var (
+		retries int
+		err     error
+	)
+	for retries < MaxRetries {
+		nsSpec := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
+		if _, err = client.Kube.CoreV1().Namespaces().
+			Create(context.Background(), nsSpec, metav1.CreateOptions{}); err == nil {
+			return nil
+		}
+		retries++
+		time.Sleep(RetrySleepDuration)
+	}
+	return err
 }
 
 // TearDown will delete created names using clients.
@@ -227,8 +297,11 @@ func TearDown(client *Client) {
 	}
 
 	client.Tracker.Clean(true)
-	if err := DeleteNameSpace(client); err != nil {
-		client.T.Logf("Could not delete the namespace %q: %v", client.Namespace, err)
+	// If we're reusing existing namespaces leave the deletion to the creator.
+	if !ReuseNamespace {
+		if err := DeleteNameSpace(client); err != nil {
+			client.T.Logf("Could not delete the namespace %q: %v", client.Namespace, err)
+		}
 	}
 }
 
@@ -253,33 +326,25 @@ func formatEvent(e *corev1.Event) string {
 	}, "\n")
 }
 
-// CreateNamespaceIfNeeded creates a new namespace if it does not exist.
-func CreateNamespaceIfNeeded(t *testing.T, client *Client, namespace string) {
-	_, err := client.Kube.CoreV1().Namespaces().Get(context.Background(), namespace, metav1.GetOptions{})
-
-	if err != nil && apierrs.IsNotFound(err) {
-		nsSpec := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
-		_, err = client.Kube.CoreV1().Namespaces().Create(context.Background(), nsSpec, metav1.CreateOptions{})
-
-		if err != nil {
-			t.Fatalf("Failed to create Namespace: %s; %v", namespace, err)
-		}
-
-		// https://github.com/kubernetes/kubernetes/issues/66689
-		// We can only start creating pods after the default ServiceAccount is created by the kube-controller-manager.
-		err = waitForServiceAccountExists(client, "default", namespace)
-		if err != nil {
-			t.Fatal("The default ServiceAccount was not created for the Namespace:", namespace)
-		}
+// SetupServiceAccount creates a new namespace if it does not exist.
+func SetupServiceAccount(t *testing.T, client *Client) {
+	// https://github.com/kubernetes/kubernetes/issues/66689
+	// We can only start creating pods after the default ServiceAccount is created by the kube-controller-manager.
+	err := waitForServiceAccountExists(client, "default", client.Namespace)
+	if err != nil {
+		t.Fatal("The default ServiceAccount was not created for the Namespace:", client.Namespace)
+	}
+}
 
-		// If the "default" Namespace has a secret called
-		// "kn-eventing-test-pull-secret" then use that as the ImagePullSecret
-		// on the "default" ServiceAccount in this new Namespace.
-		// This is needed for cases where the images are in a private registry.
-		_, err := utils.CopySecret(client.Kube.CoreV1(), "default", testPullSecretName, namespace, "default")
-		if err != nil && !apierrs.IsNotFound(err) {
-			t.Fatalf("error copying the secret into ns %q: %s", namespace, err)
-		}
+// SetupPullSecret sets up kn-eventing-test-pull-secret on the client namespace.
+func SetupPullSecret(t *testing.T, client *Client) {
+	// If the "default" Namespace has a secret called
+	// "kn-eventing-test-pull-secret" then use that as the ImagePullSecret
+	// on the "default" ServiceAccount in this new Namespace.
+	// This is needed for cases where the images are in a private registry.
+	_, err := utils.CopySecret(client.Kube.CoreV1(), "default", testPullSecretName, client.Namespace, "default")
+	if err != nil && !apierrs.IsNotFound(err) {
+		t.Fatalf("error copying the secret into ns %q: %s", client.Namespace, err)
 	}
 }
 
diff --git a/vendor/knative.dev/pkg/test/environment/config.go b/vendor/knative.dev/pkg/test/environment/config.go
index 39cfe9014..6591a4637 100644
--- a/vendor/knative.dev/pkg/test/environment/config.go
+++ b/vendor/knative.dev/pkg/test/environment/config.go
@@ -17,7 +17,7 @@ limitations under the License.
 // This file contains logic to encapsulate flags which are needed to specify
 // what cluster, etc. to use for e2e tests.
 
-package flags
+package environment
 
 import (
 	"flag"