Repository: knative-sandbox/eventing-kafka
Pull request #383 (2021-02-12 21:43:00)

slightly improve e2e doc
 

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Add skip-teardown flag instruction
- Add instruction to run the e2e mt source tests
-

<!--
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/383#" 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:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/master/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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/383?src=pr&el=h1) Report
> Merging [#383](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/383?src=pr&el=desc) (1b699af) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/5cab2571692609ab83d57d64cbbc5f1f5fe5484f?el=desc) (5cab257) will **increase** coverage by `0.14%`.
> The diff coverage is `n/a`.

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

```diff
@@            Coverage Diff             @@
##           master     #383      +/-   ##
==========================================
+ Coverage   75.12%   75.27%   +0.14%     
==========================================
  Files         124      124              
  Lines        4792     4792              
==========================================
+ Hits         3600     3607       +7     
+ Misses        969      962       -7     
  Partials      223      223              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/383?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/383/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `71.81% <0.00%> (+6.36%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/383?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/383?src=pr&el=footer). Last update [5cab257...9f150b8](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/383?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 9f150b8a52d18194e34890cca0d954526b32bef4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/383/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1360343684637265920) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

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

<details>

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

Patch:
diff --git a/config/source/multi/201-adapter-clusterrole.yaml b/config/source/multi/201-adapter-clusterrole.yaml
index dc3f471c6..43163dc3b 100644
--- a/config/source/multi/201-adapter-clusterrole.yaml
+++ b/config/source/multi/201-adapter-clusterrole.yaml
@@ -56,14 +56,27 @@ rules:
   - watch
 
 - apiGroups:
-    - ""
+  - apps
   resources:
-    - secrets
+  - deployments
   verbs:
-    - get
+  - get
+  - list
+  - watch
+  - create
+  - update
+  - patch
+  - delete
 
 - apiGroups:
-   - ""
+  - ""
+  resources:
+  - secrets
+  verbs:
+  - get
+
+- apiGroups:
+  - ""
   resources:
   - events
   verbs:
diff --git a/go.mod b/go.mod
index 1805633ff..9e527c26d 100644
--- a/go.mod
+++ b/go.mod
@@ -35,5 +35,5 @@ require (
 	k8s.io/utils v0.0.0-20200729134348-d5654de09c73
 	knative.dev/eventing v0.20.1-0.20210211204829-1c771ce4058e
 	knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
-	knative.dev/pkg v0.0.0-20210211034618-e38bb8931ffe
+	knative.dev/pkg v0.0.0-20210212203835-448ae657fb5f
 )
diff --git a/go.sum b/go.sum
index 55b2ad9de..35aea31ba 100644
--- a/go.sum
+++ b/go.sum
@@ -1252,6 +1252,8 @@ knative.dev/pkg v0.0.0-20210208175252-a02dcff9ee26 h1:PkJB1M77wYBI1UF5vp8y4fRnYQ
 knative.dev/pkg v0.0.0-20210208175252-a02dcff9ee26/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
 knative.dev/pkg v0.0.0-20210211034618-e38bb8931ffe h1:DH5UryA0fIXdNm9BYnqYo7X1IU8LlkVv8dd2ghkm8m8=
 knative.dev/pkg v0.0.0-20210211034618-e38bb8931ffe/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
+knative.dev/pkg v0.0.0-20210212203835-448ae657fb5f h1:pz+UDOTsCIClHXGM/SgCzkctLxgi3+P0C26x6Ujlqho=
+knative.dev/pkg v0.0.0-20210212203835-448ae657fb5f/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
 knative.dev/reconciler-test v0.0.0-20210210224717-0428cb00fe76/go.mod h1:hIOndGzuU1mxSJ8IF0RqPL/rAoazZZ0NjuJoMwKeNBM=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
diff --git a/pkg/common/scheduler/statefulset/autoscaler.go b/pkg/common/scheduler/statefulset/autoscaler.go
index 597e9356c..47332745f 100644
--- a/pkg/common/scheduler/statefulset/autoscaler.go
+++ b/pkg/common/scheduler/statefulset/autoscaler.go
@@ -46,7 +46,8 @@ func (s *StatefulSetScheduler) autoscale(ctx context.Context) {
 				zap.Int32("free", s.freeCapacity(snapshot)),
 				zap.Int32("used", s.usedCapacity(snapshot)),
 				zap.Int32("pending", s.pendingVReplicas()),
-				zap.Int32("replicas", s.replicas))
+				zap.Int32("replicas", s.replicas),
+				zap.Int32("last ordinal", snapshot.lastOrdinal))
 
 			var ratio float64
 			if s.replicas == 0 {
@@ -72,13 +73,21 @@ func (s *StatefulSetScheduler) autoscale(ctx context.Context) {
 				}
 
 				// Desired ratio is 0.5 (TODO: configurable)
-				scale.Spec.Replicas = int32(math.Ceil(float64(s.usedCapacity(snapshot)+s.pendingVReplicas()) / (float64(s.capacity) * 0.5)))
+				new := int32(math.Ceil(float64(s.usedCapacity(snapshot)+s.pendingVReplicas()) / (float64(s.capacity) * 0.5)))
 
-				s.logger.Infow("updating adapter replicas", zap.Int32("replicas", scale.Spec.Replicas))
+				// Make sure not to scale down past the last pod with placed vpods
+				if new < snapshot.lastOrdinal+1 {
+					new = snapshot.lastOrdinal + 1
+				}
 
-				_, err = s.statefulSetClient.UpdateScale(ctx, s.statefulSetName, scale, metav1.UpdateOptions{})
-				if err != nil {
-					s.logger.Errorw("updating scale subresource failed", zap.Error(err))
+				if new != scale.Spec.Replicas {
+					scale.Spec.Replicas = new
+					s.logger.Infow("updating adapter replicas", zap.Int32("replicas", scale.Spec.Replicas))
+
+					_, err = s.statefulSetClient.UpdateScale(ctx, s.statefulSetName, scale, metav1.UpdateOptions{})
+					if err != nil {
+						s.logger.Errorw("updating scale subresource failed", zap.Error(err))
+					}
 				}
 			}
 		}()
diff --git a/pkg/common/scheduler/statefulset/snapshot.go b/pkg/common/scheduler/statefulset/snapshot.go
index a2891c1e3..58632ba4c 100644
--- a/pkg/common/scheduler/statefulset/snapshot.go
+++ b/pkg/common/scheduler/statefulset/snapshot.go
@@ -22,6 +22,10 @@ import "go.uber.org/zap"
 type Snapshot struct {
 	// free tracks the free capacity of each pod.
 	free map[string]int32
+
+	// lastOrdinal is the ordinal index corresponding to the last statefulset replica
+	// with placed vpods.
+	lastOrdinal int32
 }
 
 func (s *StatefulSetScheduler) snapshot(logger *zap.SugaredLogger) (*Snapshot, error) {
@@ -36,6 +40,8 @@ func (s *StatefulSetScheduler) snapshot(logger *zap.SugaredLogger) (*Snapshot, e
 		free[podName] = s.capacity
 	}
 
+	last := int32(0)
+
 	for _, vpod := range vpods {
 		ps := vpod.GetPlacements()
 
@@ -49,7 +55,12 @@ func (s *StatefulSetScheduler) snapshot(logger *zap.SugaredLogger) (*Snapshot, e
 				// Over committed. The autoscaler will fix it.
 				logger.Infow("pod is overcommitted", zap.String("podName", podName))
 			}
+
+			ordinal := ordinalFromPodName(podName)
+			if ordinal > last {
+				last = ordinal
+			}
 		}
 	}
-	return &Snapshot{free: free}, nil
+	return &Snapshot{free: free, lastOrdinal: last}, nil
 }
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index 2dba25b4f..c9f2772fd 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -84,10 +84,6 @@ func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, httpMe
 }
 
 func (a *Adapter) Start(ctx context.Context) error {
-	return a.start(ctx.Done())
-}
-
-func (a *Adapter) start(stopCh <-chan struct{}) error {
 	a.logger.Infow("Starting with config: ",
 		zap.String("Topics", strings.Join(a.config.Topics, ",")),
 		zap.String("ConsumerGroup", a.config.ConsumerGroup),
@@ -107,7 +103,12 @@ func (a *Adapter) start(stopCh <-chan struct{}) error {
 	if err != nil {
 		panic(err)
 	}
-	defer func() { _ = group.Close() }()
+	defer func() {
+		err := group.Close()
+		if err != nil {
+			a.logger.Errorw("Failed to close consumer group", zap.Error(err))
+		}
+	}()
 
 	// Track errors
 	go func() {
@@ -116,7 +117,7 @@ func (a *Adapter) start(stopCh <-chan struct{}) error {
 		}
 	}()
 
-	<-stopCh
+	<-ctx.Done()
 	a.logger.Info("Shutting down...")
 	return nil
 }
diff --git a/pkg/source/mtadapter/adapter.go b/pkg/source/mtadapter/adapter.go
index 6172caeb9..7ff5c824a 100644
--- a/pkg/source/mtadapter/adapter.go
+++ b/pkg/source/mtadapter/adapter.go
@@ -90,16 +90,19 @@ func (a *Adapter) Start(ctx context.Context) error {
 // Implements MTAdapter
 
 func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
+	a.sourcesMu.Lock()
+	defer a.sourcesMu.Unlock()
+
 	key := obj.Namespace + "/" + obj.Name
 
-	a.sourcesMu.RLock()
-	cancel, ok := a.sources[key]
-	a.sourcesMu.RUnlock()
+	_, ok := a.sources[key]
 
 	if ok {
+		// TODO: see https://github.com/knative-sandbox/eventing-kafka/issues/382
+		return
 		// TODO: do not stop if the only thing that changes is the number of vreplicas
-		a.logger.Info("stopping adapter", zap.String("key", key))
-		cancel()
+		//a.logger.Info("stopping adapter", zap.String("key", key))
+		//cancel()
 	}
 
 	placement := scheduler.GetPlacementForPod(obj.GetPlacements(), a.config.PodName)
@@ -166,17 +169,16 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 
 	}(ctx)
 
-	a.sourcesMu.Lock()
 	a.sources[key] = cancelFn
-	a.sourcesMu.Unlock()
 }
 
 func (a *Adapter) Remove(ctx context.Context, obj *v1beta1.KafkaSource) {
+	a.sourcesMu.Lock()
+	defer a.sourcesMu.Unlock()
+
 	key := obj.Namespace + "/" + obj.Name
 
-	a.sourcesMu.RLock()
 	cancel, ok := a.sources[key]
-	a.sourcesMu.RUnlock()
 
 	if !ok {
 		return
@@ -184,9 +186,7 @@ func (a *Adapter) Remove(ctx context.Context, obj *v1beta1.KafkaSource) {
 
 	cancel()
 
-	a.sourcesMu.Lock()
 	delete(a.sources, key)
-	a.sourcesMu.Unlock()
 }
 
 // ResolveSecret resolves the secret reference
diff --git a/vendor/knative.dev/pkg/controller/stats_reporter.go b/vendor/knative.dev/pkg/controller/stats_reporter.go
index 92a5848a1..69ff56ea9 100644
--- a/vendor/knative.dev/pkg/controller/stats_reporter.go
+++ b/vendor/knative.dev/pkg/controller/stats_reporter.go
@@ -34,8 +34,8 @@ import (
 )
 
 var (
-	workQueueDepthStat   = stats.Int64("work_queue_depth", "Depth of the work queue", stats.UnitNone)
-	reconcileCountStat   = stats.Int64("reconcile_count", "Number of reconcile operations", stats.UnitNone)
+	workQueueDepthStat   = stats.Int64("work_queue_depth", "Depth of the work queue", stats.UnitDimensionless)
+	reconcileCountStat   = stats.Int64("reconcile_count", "Number of reconcile operations", stats.UnitDimensionless)
 	reconcileLatencyStat = stats.Int64("reconcile_latency", "Latency of reconcile operations", stats.UnitMilliseconds)
 
 	// reconcileDistribution defines the bucket boundaries for the histogram of reconcile latency metric.
@@ -58,37 +58,37 @@ func init() {
 		Adds: stats.Int64(
 			"workqueue_adds_total",
 			"Total number of adds handled by workqueue",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		Depth: stats.Int64(
 			"workqueue_depth",
 			"Current depth of workqueue",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		Latency: stats.Float64(
 			"workqueue_queue_latency_seconds",
 			"How long in seconds an item stays in workqueue before being requested.",
-			"s",
+			stats.UnitSeconds,
 		),
 		Retries: stats.Int64(
 			"workqueue_retries_total",
 			"Total number of retries handled by workqueue",
-			"s",
+			stats.UnitDimensionless,
 		),
 		WorkDuration: stats.Float64(
 			"workqueue_work_duration_seconds",
 			"How long in seconds processing an item from workqueue takes.",
-			"s",
+			stats.UnitSeconds,
 		),
 		UnfinishedWorkSeconds: stats.Float64(
 			"workqueue_unfinished_work_seconds",
 			"How long in seconds the outstanding workqueue items have been in flight (total).",
-			"s",
+			stats.UnitSeconds,
 		),
 		LongestRunningProcessorSeconds: stats.Float64(
 			"workqueue_longest_running_processor_seconds",
 			"How long in seconds the longest outstanding workqueue item has been in flight.",
-			"s",
+			stats.UnitSeconds,
 		),
 	}
 	workqueue.SetProvider(wp)
@@ -99,49 +99,49 @@ func init() {
 		ItemsInList: stats.Float64(
 			"reflector_items_in_list",
 			"How many items an API list returns to the reflectors",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		// TODO(mattmoor): This is not in the latest version, so it will
 		// be removed in a future version.
 		ItemsInMatch: stats.Float64(
 			"reflector_items_in_match",
 			"",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		ItemsInWatch: stats.Float64(
 			"reflector_items_in_watch",
 			"How many items an API watch returns to the reflectors",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		LastResourceVersion: stats.Float64(
 			"reflector_last_resource_version",
 			"Last resource version seen for the reflectors",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		ListDuration: stats.Float64(
 			"reflector_list_duration_seconds",
 			"How long an API list takes to return and decode for the reflectors",
-			stats.UnitNone,
+			stats.UnitSeconds,
 		),
 		Lists: stats.Int64(
 			"reflector_lists_total",
 			"Total number of API lists done by the reflectors",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		ShortWatches: stats.Int64(
 			"reflector_short_watches_total",
 			"Total number of short API watches done by the reflectors",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		WatchDuration: stats.Float64(
 			"reflector_watch_duration_seconds",
 			"How long an API watch takes to return and decode for the reflectors",
-			stats.UnitNone,
+			stats.UnitSeconds,
 		),
 		Watches: stats.Int64(
 			"reflector_watches_total",
 			"Total number of API watches done by the reflectors",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 	}
 	cache.SetReflectorMetricsProvider(rp)
@@ -150,12 +150,12 @@ func init() {
 		Latency: stats.Float64(
 			"client_latency",
 			"How long Kubernetes API requests take",
-			"s",
+			stats.UnitSeconds,
 		),
 		Result: stats.Int64(
 			"client_results",
 			"Total number of API requests (broken down by status code)",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 	}
 	opts := kubemetrics.RegisterOpts{
diff --git a/vendor/knative.dev/pkg/metrics/memstats.go b/vendor/knative.dev/pkg/metrics/memstats.go
index 719c7e018..d96a0f4bc 100644
--- a/vendor/knative.dev/pkg/metrics/memstats.go
+++ b/vendor/knative.dev/pkg/metrics/memstats.go
@@ -33,112 +33,112 @@ func NewMemStatsAll() *MemStatsProvider {
 		Alloc: stats.Int64(
 			"go_alloc",
 			"The number of bytes of allocated heap objects.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		TotalAlloc: stats.Int64(
 			"go_total_alloc",
 			"The cumulative bytes allocated for heap objects.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		Sys: stats.Int64(
 			"go_sys",
 			"The total bytes of memory obtained from the OS.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		Lookups: stats.Int64(
 			"go_lookups",
 			"The number of pointer lookups performed by the runtime.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		Mallocs: stats.Int64(
 			"go_mallocs",
 			"The cumulative count of heap objects allocated.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		Frees: stats.Int64(
 			"go_frees",
 			"The cumulative count of heap objects freed.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		HeapAlloc: stats.Int64(
 			"go_heap_alloc",
 			"The number of bytes of allocated heap objects.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		HeapSys: stats.Int64(
 			"go_heap_sys",
 			"The number of bytes of heap memory obtained from the OS.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		HeapIdle: stats.Int64(
 			"go_heap_idle",
 			"The number of bytes in idle (unused) spans.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		HeapInuse: stats.Int64(
 			"go_heap_in_use",
 			"The number of bytes in in-use spans.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		HeapReleased: stats.Int64(
 			"go_heap_released",
 			"The number of bytes of physical memory returned to the OS.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		HeapObjects: stats.Int64(
 			"go_heap_objects",
 			"The number of allocated heap objects.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		StackInuse: stats.Int64(
 			"go_stack_in_use",
 			"The number of bytes in stack spans.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		StackSys: stats.Int64(
 			"go_stack_sys",
 			"The number of bytes of stack memory obtained from the OS.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		MSpanInuse: stats.Int64(
 			"go_mspan_in_use",
 			"The number of bytes of allocated mspan structures.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		MSpanSys: stats.Int64(
 			"go_mspan_sys",
 			"The number of bytes of memory obtained from the OS for mspan structures.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		MCacheInuse: stats.Int64(
 			"go_mcache_in_use",
 			"The number of bytes of allocated mcache structures.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		MCacheSys: stats.Int64(
 			"go_mcache_sys",
 			"The number of bytes of memory obtained from the OS for mcache structures.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		BuckHashSys: stats.Int64(
 			"go_bucket_hash_sys",
 			"The number of bytes of memory in profiling bucket hash tables.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		GCSys: stats.Int64(
 			"go_gc_sys",
 			"The number of bytes of memory in garbage collection metadata.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		OtherSys: stats.Int64(
 			"go_other_sys",
 			"The number of bytes of memory in miscellaneous off-heap runtime allocations.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		NextGC: stats.Int64(
 			"go_next_gc",
 			"The target heap size of the next GC cycle.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		LastGC: stats.Int64(
 			"go_last_gc",
@@ -153,17 +153,17 @@ func NewMemStatsAll() *MemStatsProvider {
 		NumGC: stats.Int64(
 			"go_num_gc",
 			"The number of completed GC cycles.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		NumForcedGC: stats.Int64(
 			"go_num_forced_gc",
 			"The number of GC cycles that were forced by the application calling the GC function.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 		GCCPUFraction: stats.Float64(
 			"go_gc_cpu_fraction",
 			"The fraction of this program's available CPU time used by the GC since the program started.",
-			stats.UnitNone,
+			stats.UnitDimensionless,
 		),
 	}
 }
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 4e9c6ef97..67b301546 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1218,7 +1218,7 @@ knative.dev/eventing/test/test_images/recordevents
 # knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20210211034618-e38bb8931ffe
+# knative.dev/pkg v0.0.0-20210212203835-448ae657fb5f
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/pkg/common/scheduler/statefulset/snapshot_test.go b/pkg/common/scheduler/statefulset/snapshot_test.go
index f35ce2a82..73170a534 100644
--- a/pkg/common/scheduler/statefulset/snapshot_test.go
+++ b/pkg/common/scheduler/statefulset/snapshot_test.go
@@ -45,19 +45,19 @@ func TestSnapshot(t *testing.T) {
 			name:     "no vpods, no replicas",
 			vpods:    [][]duckv1alpha1.Placement{},
 			replicas: int32(0),
-			expected: Snapshot{free: map[string]int32{}},
+			expected: Snapshot{free: map[string]int32{}, lastOrdinal: 0},
 		},
 		{
 			name:     "no vpods, one replica",
 			vpods:    [][]duckv1alpha1.Placement{},
 			replicas: int32(1),
-			expected: Snapshot{free: map[string]int32{"statefulset-name-0": int32(10)}},
+			expected: Snapshot{free: map[string]int32{"statefulset-name-0": int32(10)}, lastOrdinal: 0},
 		},
 		{
 			name:     "one vpods, one replica",
 			vpods:    [][]duckv1alpha1.Placement{{{PodName: "statefulset-name-0", VReplicas: 1}}},
 			replicas: int32(1),
-			expected: Snapshot{free: map[string]int32{"statefulset-name-0": int32(9)}},
+			expected: Snapshot{free: map[string]int32{"statefulset-name-0": int32(9)}, lastOrdinal: 0},
 		},
 		{
 			name: "many vpods, many replicas",
@@ -71,7 +71,7 @@ func TestSnapshot(t *testing.T) {
 				"statefulset-name-0": int32(8),
 				"statefulset-name-1": int32(5),
 				"statefulset-name-2": int32(5),
-				"statefulset-name-3": int32(10)}},
+				"statefulset-name-3": int32(10)}, lastOrdinal: 2},
 		},
 	}
 
diff --git a/test/README.md b/test/README.md
index e2bc37b04..74c148a68 100644
--- a/test/README.md
+++ b/test/README.md
@@ -57,6 +57,8 @@ If you have set up a running environment that meets
 [the e2e test environment requirements](#environment-requirements), you can run
 with `test/e2e-tests.sh --run-tests --skip-knative-setup`.
 
+If you want to skip tearing down your cluster, add the `--skip-teardowns` flag.
+
 ## Running tests with `go test` command
 
 ### Running unit tests
@@ -98,6 +100,13 @@ bevavior you can specify a different tag through `-tag`:
 SYSTEM_NAMESPACE=knative-eventing go test -v -tags=e2e,source -count=1 ./test/e2e -tag e2e
 ```
 
+To run the e2e source tests against the multi-tenant source adapter you need to
+specify the build tags `source` and `mtsource`:
+
+```bash
+SYSTEM_NAMESPACE=knative-eventing go test -v -tags=source,mtsource  ./test/e2e
+```
+
 #### One test case
 
 To run one e2e test case, e.g. `TestSingleBinaryEventForChannel`, use
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index f55a1ae35..b53337e00 100755
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -456,7 +456,7 @@ function test_mt_source() {
   echo "Testing the multi-tenant source"
   install_mt_source || return 1
 
-  go_test_e2e -tags=source,mtsource -timeout=5m -test.parallel=${TEST_PARALLEL} ./test/...  || fail_test
+  go_test_e2e -tags=source,mtsource -timeout=20m -test.parallel=${TEST_PARALLEL} ./test/e2e/...  || fail_test
 
   uninstall_mt_source || return 1
 }