Repository: knative-sandbox/eventing-kafka
Pull request #348 (2021-02-01 14:06:06)

[master] Upgrade to latest dependencies
Cron. -knative-prow-robot

/cc knative-sandbox/channel-wg-leads
/assign knative-sandbox/channel-wg-leads

Produced via:
```shell
./hack/update-deps.sh --upgrade --release v0.21
./hack/update-codegen.sh
```

Details:
```
bumping knative.dev/eventing b20c96b...321a839:
  > 321a839 Format markdown (# 4823)
  > ec63881 Fix spec # 4515 (# 4654)
  > 1528750 upgrade to latest dependencies (# 4818)
  > 9dc57e2 Update common github actions (# 4817)
```

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/348#" title="Author self-approved">knative-automation</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **davyodom** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @davyodom` in a comment when ready.

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

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["davyodom"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/348?src=pr&el=h1) Report
> Merging [#348](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/348?src=pr&el=desc) (2a0e37e) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/ca62759ec2cdc23c391859b308eb4dadf56c107d?el=desc) (ca62759) will **not change** coverage.
> The diff coverage is `n/a`.

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

```diff
@@           Coverage Diff           @@
##           master     #348   +/-   ##
=======================================
  Coverage   76.06%   76.06%           
=======================================
  Files         114      114           
  Lines        4425     4425           
=======================================
  Hits         3366     3366           
  Misses        860      860           
  Partials      199      199           
```



------

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


Patch:
diff --git a/OWNERS b/OWNERS
index 96e56362a..78fee8de6 100644
--- a/OWNERS
+++ b/OWNERS
@@ -18,6 +18,7 @@ reviewers:
 - matzew
 - lberk
 - slinkydeveloper
+- steven0711dong
 - davyodom
 - travis-minke-sap
 - phamilton
diff --git a/cmd/source/mt_controller/main.go b/cmd/source/mt_controller/main.go
new file mode 100644
index 000000000..e81e891d8
--- /dev/null
+++ b/cmd/source/mt_controller/main.go
@@ -0,0 +1,203 @@
+/*
+Copyright 2019 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 main
+
+import (
+	"context"
+	"os"
+
+	"knative.dev/eventing-kafka/pkg/apis/bindings"
+	"knative.dev/eventing-kafka/pkg/apis/sources"
+	"knative.dev/pkg/webhook/resourcesemantics/conversion"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	bindingsv1alpha1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1alpha1"
+	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	sourcesv1alpha1 "knative.dev/eventing-kafka/pkg/apis/sources/v1alpha1"
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+
+	"knative.dev/eventing-kafka/pkg/source/reconciler/binding"
+	source "knative.dev/eventing-kafka/pkg/source/reconciler/mtsource"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/injection"
+	"knative.dev/pkg/injection/sharedmain"
+	"knative.dev/pkg/signals"
+	"knative.dev/pkg/webhook"
+	"knative.dev/pkg/webhook/certificates"
+	"knative.dev/pkg/webhook/psbinding"
+	"knative.dev/pkg/webhook/resourcesemantics"
+	"knative.dev/pkg/webhook/resourcesemantics/defaulting"
+	"knative.dev/pkg/webhook/resourcesemantics/validation"
+)
+
+const (
+	component = "kafka-controller"
+)
+
+var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
+	// v1alpha1
+	sourcesv1alpha1.SchemeGroupVersion.WithKind("KafkaSource"):   &sourcesv1alpha1.KafkaSource{},
+	bindingsv1alpha1.SchemeGroupVersion.WithKind("KafkaBinding"): &bindingsv1alpha1.KafkaBinding{},
+	// v1beta1
+	sourcesv1beta1.SchemeGroupVersion.WithKind("KafkaSource"):   &sourcesv1beta1.KafkaSource{},
+	bindingsv1beta1.SchemeGroupVersion.WithKind("KafkaBinding"): &bindingsv1beta1.KafkaBinding{},
+}
+
+var callbacks = map[schema.GroupVersionKind]validation.Callback{}
+
+func NewDefaultingAdmissionController(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
+	return defaulting.NewAdmissionController(ctx,
+
+		// Name of the resource webhook.
+		"defaulting.webhook.kafka.sources.knative.dev",
+
+		// The path on which to serve the webhook.
+		"/defaulting",
+
+		// The resources to default.
+		types,
+
+		// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
+		func(ctx context.Context) context.Context {
+			// Here is where you would infuse the context with state
+			// (e.g. attach a store with configmap data)
+			return ctx
+		},
+
+		// Whether to disallow unknown fields.
+		true,
+	)
+}
+
+func NewValidationAdmissionController(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
+	return validation.NewAdmissionController(ctx,
+
+		// Name of the resource webhook.
+		"validation.webhook.kafka.sources.knative.dev",
+
+		// The path on which to serve the webhook.
+		"/resource-validation",
+
+		// The resources to validate.
+		types,
+
+		// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
+		func(ctx context.Context) context.Context {
+			// Here is where you would infuse the context with state
+			// (e.g. attach a store with configmap data)
+			return ctx
+		},
+
+		// Whether to disallow unknown fields.
+		true,
+
+		// Extra validating callbacks to be applied to resources.
+		callbacks,
+	)
+}
+
+func NewKafkaBindingWebhook(opts ...psbinding.ReconcilerOption) injection.ControllerConstructor {
+	return func(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
+		return psbinding.NewAdmissionController(ctx,
+
+			// Name of the resource webhook.
+			"kafkabindings.webhook.kafka.sources.knative.dev",
+
+			// The path on which to serve the webhook.
+			"/kafkabindings",
+
+			// How to get all the Bindables for configuring the mutating webhook.
+			binding.ListAll,
+
+			// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
+			func(ctx context.Context, _ psbinding.Bindable) (context.Context, error) {
+				// Here is where you would infuse the context with state
+				// (e.g. attach a store with configmap data)
+				return ctx, nil
+			},
+			opts...,
+		)
+	}
+}
+
+func NewConversionController(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
+	var (
+		sourcesv1alpha1_  = sourcesv1alpha1.SchemeGroupVersion.Version
+		bindingsv1alpha1_ = bindingsv1alpha1.SchemeGroupVersion.Version
+		sourcesv1beta1_   = sourcesv1beta1.SchemeGroupVersion.Version
+		bindingsv1beta1_  = bindingsv1beta1.SchemeGroupVersion.Version
+	)
+
+	return conversion.NewConversionController(ctx,
+		// The path on which to serve the webhook
+		"/resource-conversion",
+
+		// Specify the types of custom resource definitions that should be converted
+		map[schema.GroupKind]conversion.GroupKindConversion{
+			// KafkaSource
+			sourcesv1beta1.Kind("KafkaSource"): {
+				DefinitionName: sources.KafkaSourcesResource.String(),
+				HubVersion:     sourcesv1alpha1_,
+				Zygotes: map[string]conversion.ConvertibleObject{
+					sourcesv1alpha1_: &sourcesv1alpha1.KafkaSource{},
+					sourcesv1beta1_:  &sourcesv1beta1.KafkaSource{},
+				},
+			},
+			// KafkaBinding
+			bindingsv1beta1.Kind("KafkaBinding"): {
+				DefinitionName: bindings.KafkaBindingsResource.String(),
+				HubVersion:     bindingsv1alpha1_,
+				Zygotes: map[string]conversion.ConvertibleObject{
+					bindingsv1alpha1_: &bindingsv1alpha1.KafkaBinding{},
+					bindingsv1beta1_:  &bindingsv1beta1.KafkaBinding{},
+				},
+			},
+		},
+
+		// A function that infuses the context passed to ConvertTo/ConvertFrom/SetDefaults with custom metadata.
+		func(ctx context.Context) context.Context {
+			return ctx
+		},
+	)
+}
+
+func main() {
+	ctx := webhook.WithOptions(signals.NewContext(), webhook.Options{
+		ServiceName: "kafka-source-webhook",
+		Port:        8443,
+		SecretName:  "kafka-source-webhook-certs",
+	})
+
+	kfkSelector := psbinding.WithSelector(psbinding.ExclusionSelector)
+	if os.Getenv("KAFKA_BINDING_SELECTION_MODE") == "inclusion" {
+		kfkSelector = psbinding.WithSelector(psbinding.InclusionSelector)
+	}
+
+	sharedmain.WebhookMainWithContext(ctx, component,
+		certificates.NewController,
+		NewDefaultingAdmissionController,
+		NewValidationAdmissionController,
+		NewConversionController,
+
+		// For each binding we have a controller and a binding webhook.
+		binding.NewController, NewKafkaBindingWebhook(kfkSelector),
+
+		source.NewController,
+	)
+}
diff --git a/cmd/source/mt_receive_adapter/main.go b/cmd/source/mt_receive_adapter/main.go
new file mode 100644
index 000000000..57c7362ef
--- /dev/null
+++ b/cmd/source/mt_receive_adapter/main.go
@@ -0,0 +1,31 @@
+/*
+Copyright 2019 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 main
+
+import (
+	"knative.dev/pkg/signals"
+
+	"knative.dev/eventing-kafka/pkg/source/mtadapter"
+	"knative.dev/eventing/pkg/adapter/v2"
+)
+
+func main() {
+	ctx := signals.NewContext()
+
+	ctx = adapter.WithController(ctx, mtadapter.NewController)
+	adapter.MainWithContext(ctx, "kafkasource", mtadapter.NewEnvConfig, mtadapter.NewAdapter)
+}
diff --git a/config/source/multi/200-adapter-serviceaccount.yaml b/config/source/multi/200-adapter-serviceaccount.yaml
new file mode 100644
index 000000000..36ea84cbb
--- /dev/null
+++ b/config/source/multi/200-adapter-serviceaccount.yaml
@@ -0,0 +1,21 @@
+# Copyright 2019 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.
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: kafkasource-mt-adapter
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
diff --git a/config/source/multi/200-serviceaccount.yaml b/config/source/multi/200-serviceaccount.yaml
new file mode 100644
index 000000000..87af514b8
--- /dev/null
+++ b/config/source/multi/200-serviceaccount.yaml
@@ -0,0 +1,21 @@
+# Copyright 2019 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.
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: kafka-controller-manager
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
diff --git a/config/source/multi/201-adapter-clusterrole.yaml b/config/source/multi/201-adapter-clusterrole.yaml
new file mode 100644
index 000000000..367d62d6f
--- /dev/null
+++ b/config/source/multi/201-adapter-clusterrole.yaml
@@ -0,0 +1,68 @@
+# Copyright 2019 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.
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: eventing-sources-kafka-adapter
+  labels:
+    kafka.eventing.knative.dev/release: devel
+rules:
+- apiGroups:
+  - sources.knative.dev
+  resources:
+  - kafkasources
+  verbs:
+  - get
+  - list
+  - watch
+  - patch
+
+- apiGroups:
+  - sources.knative.dev
+  resources:
+  - kafkasources/status
+  verbs:
+  - get
+
+- apiGroups:
+  - sources.knative.dev
+  resources:
+  - kafkasources/finalizers
+  verbs:
+  - get
+  - patch
+  - delete
+
+- apiGroups:
+  - ""
+  resources:
+  - events
+  - configmaps
+  verbs:
+  - get
+  - list
+  - watch
+
+- apiGroups:
+   - ""
+  resources:
+  - events
+  verbs:
+  - get
+  - list
+  - watch
+  - patch
+  - create
+  - update
diff --git a/config/source/multi/201-clusterrole.yaml b/config/source/multi/201-clusterrole.yaml
new file mode 120000
index 000000000..7b9e92d83
--- /dev/null
+++ b/config/source/multi/201-clusterrole.yaml
@@ -0,0 +1,1 @@
+roles/clusterrole.yaml
\ No newline at end of file
diff --git a/config/source/multi/202-adapter-clusterrolebinding.yaml b/config/source/multi/202-adapter-clusterrolebinding.yaml
new file mode 100644
index 000000000..5ea2bf798
--- /dev/null
+++ b/config/source/multi/202-adapter-clusterrolebinding.yaml
@@ -0,0 +1,28 @@
+# Copyright 2019 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.
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: eventing-sources-kafka-adapter
+  labels:
+    kafka.eventing.knative.dev/release: devel
+subjects:
+- kind: ServiceAccount
+  name: kafkasource-mt-adapter
+  namespace: knative-eventing
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: eventing-sources-kafka-adapter
diff --git a/config/source/multi/202-clusterrolebinding.yaml b/config/source/multi/202-clusterrolebinding.yaml
new file mode 100644
index 000000000..e04c8977c
--- /dev/null
+++ b/config/source/multi/202-clusterrolebinding.yaml
@@ -0,0 +1,66 @@
+# Copyright 2019 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.
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: eventing-sources-kafka-controller
+  labels:
+    kafka.eventing.knative.dev/release: devel
+subjects:
+- kind: ServiceAccount
+  name: kafka-controller-manager
+  namespace: knative-eventing
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: eventing-sources-kafka-controller
+
+---
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: eventing-sources-kafka-controller-addressable-resolver
+  labels:
+    kafka.eventing.knative.dev/release: devel
+subjects:
+- kind: ServiceAccount
+  name: kafka-controller-manager
+  namespace: knative-eventing
+# An aggregated ClusterRole for all Addressable CRDs.
+# Ref: https://github.com/knative/eventing/tree/master/config/core/rolesaddressable-resolvers-clusterrole.yaml
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: addressable-resolver
+
+---
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: eventing-sources-kafka-controller-podspecable-binding
+  labels:
+    kafka.eventing.knative.dev/release: devel
+subjects:
+- kind: ServiceAccount
+  name: kafka-controller-manager
+  namespace: knative-eventing
+# An aggregated ClusterRole for all PodSpecable CRDs.
+# Ref: https://github.com/knative/eventing/blob/master/config/core/roles/podspecable-binding-clusterrole.yaml
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: podspecable-binding
diff --git a/config/source/multi/300-kafkabinding.yaml b/config/source/multi/300-kafkabinding.yaml
new file mode 120000
index 000000000..627db3a76
--- /dev/null
+++ b/config/source/multi/300-kafkabinding.yaml
@@ -0,0 +1,1 @@
+resources/kafkabinding.yaml
\ No newline at end of file
diff --git a/config/source/multi/300-kafkasource.yaml b/config/source/multi/300-kafkasource.yaml
new file mode 120000
index 000000000..a2524b658
--- /dev/null
+++ b/config/source/multi/300-kafkasource.yaml
@@ -0,0 +1,1 @@
+resources/kafkasource.yaml
\ No newline at end of file
diff --git a/config/source/multi/500-adapter.yaml b/config/source/multi/500-adapter.yaml
new file mode 120000
index 000000000..9dfdf5245
--- /dev/null
+++ b/config/source/multi/500-adapter.yaml
@@ -0,0 +1,1 @@
+deployments/adapter.yaml
\ No newline at end of file
diff --git a/config/source/multi/500-controller-service.yaml b/config/source/multi/500-controller-service.yaml
new file mode 120000
index 000000000..462d68f19
--- /dev/null
+++ b/config/source/multi/500-controller-service.yaml
@@ -0,0 +1,1 @@
+deployments/controller-service.yaml
\ No newline at end of file
diff --git a/config/source/multi/500-controller.yaml b/config/source/multi/500-controller.yaml
new file mode 120000
index 000000000..7a5f7f74a
--- /dev/null
+++ b/config/source/multi/500-controller.yaml
@@ -0,0 +1,1 @@
+deployments/controller.yaml
\ No newline at end of file
diff --git a/config/source/multi/500-webhook-configuration.yaml b/config/source/multi/500-webhook-configuration.yaml
new file mode 120000
index 000000000..53eb87364
--- /dev/null
+++ b/config/source/multi/500-webhook-configuration.yaml
@@ -0,0 +1,1 @@
+webhooks/kafka.yaml
\ No newline at end of file
diff --git a/config/source/multi/deployments/adapter.yaml b/config/source/multi/deployments/adapter.yaml
new file mode 100644
index 000000000..85991b252
--- /dev/null
+++ b/config/source/multi/deployments/adapter.yaml
@@ -0,0 +1,71 @@
+# Copyright 2020 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.
+
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: kafkasource-mt-adapter
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
+spec:
+  replicas: 1 # TODO: set to 0
+  selector:
+    matchLabels: &labels
+      control-plane: kafkasource-mt-adapter
+  serviceName: kafkasource-mt-adapter
+  template:
+    metadata:
+      labels: *labels
+    spec:
+      serviceAccountName: kafkasource-mt-adapter
+      containers:
+      - image: ko://knative.dev/eventing-kafka/cmd/source/mt_receive_adapter
+        name: receive-adapter
+        env:
+        - name: SYSTEM_NAMESPACE
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.namespace
+        - name: POD_NAME
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.name
+
+        # DO NOT MODIFY: The values below are being filled by the kafka source controller
+        # See 500-controller.yaml
+        - name: K_METRICS_CONFIG
+          value: '{"Domain":"knative.dev/eventing","Component":"kafkasource","ConfigMap":{"profiling.enable": "true"}}'
+        - name: K_LOGGING_CONFIG
+          value: '{}'
+        - name: K_LEADER_ELECTION_CONFIG
+          value: ''
+        - name: METRICS_DOMAIN
+          value: knative.dev/sources
+
+        resources:
+          requests:
+            cpu: 50m
+            memory: 1Gi
+          limits:
+            cpu: 50m
+            memory: 1Gi
+
+        ports:
+        - name: metrics
+          containerPort: 9090
+        - name: profiling
+          containerPort: 8008
+
+      terminationGracePeriodSeconds: 10
diff --git a/config/source/multi/deployments/controller-service.yaml b/config/source/multi/deployments/controller-service.yaml
new file mode 100644
index 000000000..c95cb1a41
--- /dev/null
+++ b/config/source/multi/deployments/controller-service.yaml
@@ -0,0 +1,28 @@
+# Copyright 2020 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.
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: kafka-controller
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
+    control-plane: kafka-controller-manager
+spec:
+  selector:
+    control-plane: kafka-controller-manager
+  ports:
+  - name: https-kafka
+    port: 443
diff --git a/config/source/multi/deployments/controller.yaml b/config/source/multi/deployments/controller.yaml
new file mode 100644
index 000000000..7dc83ad54
--- /dev/null
+++ b/config/source/multi/deployments/controller.yaml
@@ -0,0 +1,83 @@
+# Copyright 2020 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.
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: kafka-controller-manager
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
+    control-plane: kafka-controller-manager
+spec:
+  replicas: 1
+  selector:
+    matchLabels: &labels
+      control-plane: kafka-controller-manager
+  template:
+    metadata:
+      labels: *labels
+    spec:
+      serviceAccountName: kafka-controller-manager
+      containers:
+      - name: manager
+        image: ko://knative.dev/eventing-kafka/cmd/source/mt_controller
+        env:
+        - name: SYSTEM_NAMESPACE
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.namespace
+        - name: METRICS_DOMAIN
+          value: knative.dev/sources
+        - name: CONFIG_OBSERVABILITY_NAME
+          value: config-observability
+        - name: CONFIG_LEADERELECTION_NAME
+          value: config-leader-election
+        volumeMounts:
+        resources:
+          requests:
+            cpu: 20m
+            memory: 20Mi
+
+        readinessProbe: &probe
+          periodSeconds: 1
+          httpGet:
+            scheme: HTTPS
+            port: 8443
+            httpHeaders:
+            - name: k-kubelet-probe
+              value: "webhook"
+        livenessProbe:
+          <<: *probe
+          initialDelaySeconds: 20
+
+      serviceAccount: kafka-controller-manager
+      terminationGracePeriodSeconds: 10
+
+---
+apiVersion: v1
+kind: Service
+metadata:
+  labels:
+    role: webhook
+    kafka.eventing.knative.dev/release: devel
+  name: kafka-source-webhook
+  namespace: knative-eventing
+spec:
+  ports:
+  - name: https-webhook
+    port: 443
+    targetPort: 8443
+  selector:
+    control-plane: kafka-controller-manager
diff --git a/config/source/multi/resources/kafkabinding.yaml b/config/source/multi/resources/kafkabinding.yaml
new file mode 100644
index 000000000..bf45b43dd
--- /dev/null
+++ b/config/source/multi/resources/kafkabinding.yaml
@@ -0,0 +1,76 @@
+# Copyright 2019 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.
+
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  labels:
+    kafka.eventing.knative.dev/release: devel
+    duck.knative.dev/binding: "true"
+    knative.dev/crd-install: "true"
+  name: kafkabindings.bindings.knative.dev
+spec:
+  group: bindings.knative.dev
+  versions:
+    - &version
+      name: v1alpha1
+      served: true
+      storage: true
+      subresources:
+        status: { }
+      schema:
+        openAPIV3Schema: &openAPIV3Schema
+          type: object
+          # this is a work around so we don't need to flush out the
+          # schema for each version at this time
+          #
+          # see issue: https://github.com/knative/serving/issues/912
+          x-kubernetes-preserve-unknown-fields: true
+      additionalPrinterColumns:
+        - name: BootstrapServers
+          type: string
+          jsonPath: ".spec.bootstrapServers"
+        - name: Ready
+          type: string
+          jsonPath: ".status.conditions[?(@.type==\"Ready\")].status"
+        - name: Reason
+          type: string
+          jsonPath: ".status.conditions[?(@.type==\"Ready\")].reason"
+        - name: Age
+          type: date
+          jsonPath: .metadata.creationTimestamp
+    - <<: *version
+      name: v1beta1
+      served: true
+      storage: false
+      schema:
+        openAPIV3Schema:
+          <<: *openAPIV3Schema
+  names:
+    categories:
+      - all
+      - knative
+      - eventing
+      - bindings
+    kind: KafkaBinding
+    plural: kafkabindings
+  scope: Namespaced
+  conversion:
+    strategy: Webhook
+    webhook:
+      conversionReviewVersions: [ "v1", "v1beta1" ]
+      clientConfig:
+        service:
+          name: kafka-source-webhook
+          namespace: knative-eventing
diff --git a/config/source/multi/resources/kafkasource.yaml b/config/source/multi/resources/kafkasource.yaml
new file mode 100644
index 000000000..a0667e64b
--- /dev/null
+++ b/config/source/multi/resources/kafkasource.yaml
@@ -0,0 +1,90 @@
+# Copyright 2019 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.
+
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  labels:
+    kafka.eventing.knative.dev/release: devel
+    eventing.knative.dev/source: "true"
+    duck.knative.dev/source: "true"
+    knative.dev/crd-install: "true"
+  annotations:
+    registry.knative.dev/eventTypes: |
+      [
+        { "type": "dev.knative.kafka.event" }
+      ]
+  name: kafkasources.sources.knative.dev
+spec:
+  group: sources.knative.dev
+  versions:
+    - &version
+      name: v1alpha1
+      served: true
+      storage: true
+      schema:
+        openAPIV3Schema: &openAPIV3Schema
+          type: object
+          # this is a work around so we don't need to flush out the
+          # schema for each version at this time
+          #
+          # see issue: https://github.com/knative/serving/issues/912
+          x-kubernetes-preserve-unknown-fields: true
+      subresources:
+        status: { }
+        scale:
+          # specReplicasPath defines the JSONPath inside of a custom resource that corresponds to Scale.Spec.Replicas.
+          specReplicasPath: .spec.consumers
+          # statusReplicasPath defines the JSONPath inside of a custom resource that corresponds to Scale.Status.Replicas.
+          statusReplicasPath: .status.consumers
+      additionalPrinterColumns:
+        - name: Topics
+          type: string
+          jsonPath: ".spec.topics"
+        - name: BootstrapServers
+          type: string
+          jsonPath: ".spec.bootstrapServers"
+        - name: Ready
+          type: string
+          jsonPath: ".status.conditions[?(@.type==\"Ready\")].status"
+        - name: Reason
+          type: string
+          jsonPath: ".status.conditions[?(@.type==\"Ready\")].reason"
+        - name: Age
+          type: date
+          jsonPath: .metadata.creationTimestamp
+    - <<: *version
+      name: v1beta1
+      served: true
+      storage: false
+      schema:
+        openAPIV3Schema:
+          <<: *openAPIV3Schema
+  names:
+    categories:
+      - all
+      - knative
+      - eventing
+      - sources
+    kind: KafkaSource
+    plural: kafkasources
+  scope: Namespaced
+  conversion:
+    strategy: Webhook
+    webhook:
+      conversionReviewVersions: [ "v1", "v1beta1" ]
+      clientConfig:
+        service:
+          name: kafka-source-webhook
+          namespace: knative-eventing
diff --git a/config/source/multi/roles/clusterrole.yaml b/config/source/multi/roles/clusterrole.yaml
new file mode 100644
index 000000000..d19bef947
--- /dev/null
+++ b/config/source/multi/roles/clusterrole.yaml
@@ -0,0 +1,166 @@
+# Copyright 2019 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.
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: eventing-sources-kafka-controller
+  labels:
+    kafka.eventing.knative.dev/release: devel
+rules:
+
+- apiGroups:
+    - sources.knative.dev
+  resources:
+    - kafkasources
+    - kafkasources/finalizers
+  verbs: &everything
+    - get
+    - list
+    - watch
+    - create
+    - update
+    - patch
+    - delete
+
+- apiGroups:
+    - sources.knative.dev
+  resources:
+    - kafkasources/status
+  verbs:
+    - get
+    - update
+    - patch
+
+
+- apiGroups:
+    - bindings.knative.dev
+  resources:
+    - kafkabindings
+    - kafkabindings/finalizers
+  verbs: &everything
+    - get
+    - list
+    - watch
+    - create
+    - update
+    - patch
+    - delete
+
+- apiGroups:
+    - bindings.knative.dev
+  resources:
+    - kafkabindings/status
+  verbs:
+    - get
+    - update
+    - patch
+
+
+- apiGroups:
+    - apps
+  resources:
+    - deployments
+    - statefulsets
+  verbs: *everything
+
+
+- apiGroups:
+    - apps
+  resources:
+    - statefulsets/scale
+  verbs:
+    - get
+    - update
+
+- apiGroups:
+    - ""
+  resources:
+    - pods
+    - services
+    - endpoints
+    - events
+    - configmaps
+    - secrets
+  verbs: *everything
+
+# let the webhook label the appropriate namespace
+- apiGroups:
+    - ""
+  resources:
+    - namespaces
+  verbs:
+    - get
+    - create
+    - update
+    - list
+    - watch
+    - patch
+
+  # For leader election
+- apiGroups:
+    - "coordination.k8s.io"
+  resources:
+    - leases
+  verbs: *everything
+
+# For actually registering our webhook.
+- apiGroups:
+    - "admissionregistration.k8s.io"
+  resources:
+    - "mutatingwebhookconfigurations"
+    - "validatingwebhookconfigurations"
+  verbs: &everything
+    - "get"
+    - "list"
+    - "create"
+    - "update"
+    - "delete"
+    - "patch"
+    - "watch"
+
+# Necessary for conversion webhook. These are copied from the serving
+# TODO: Do we really need all these permissions?
+- apiGroups:
+    - "apiextensions.k8s.io"
+  resources:
+    - "customresourcedefinitions"
+  verbs:
+    - "get"
+    - "list"
+    - "create"
+    - "update"
+    - "delete"
+    - "patch"
+    - "watch"
+
+---
+# The role is needed for the aggregated role source-observer in knative-eventing to provide readonly access to "Sources".
+# Ref: https://github.com/knative/eventing/tree/master/config/core/rolessource-observer-clusterrole.yaml.
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: eventing-kafka-source-observer
+  labels:
+    kafka.eventing.knative.dev/release: devel
+    duck.knative.dev/source: "true"
+rules:
+  - apiGroups:
+      - "sources.knative.dev"
+    resources:
+      - "kafkasources"
+    verbs:
+      - get
+      - list
+      - watch
diff --git a/config/source/multi/webhooks/kafka.yaml b/config/source/multi/webhooks/kafka.yaml
new file mode 100644
index 000000000..7ec9ee3b7
--- /dev/null
+++ b/config/source/multi/webhooks/kafka.yaml
@@ -0,0 +1,90 @@
+# Copyright 2020 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
+#
+#     https://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.
+
+apiVersion: admissionregistration.k8s.io/v1
+kind: MutatingWebhookConfiguration
+metadata:
+  name: defaulting.webhook.kafka.sources.knative.dev
+  labels:
+    kafka.eventing.knative.dev/release: devel
+webhooks:
+- admissionReviewVersions: ["v1", "v1beta1"]
+  clientConfig:
+    service:
+      name: kafka-source-webhook
+      namespace: knative-eventing
+  sideEffects: None
+  failurePolicy: Fail
+  name: defaulting.webhook.kafka.sources.knative.dev
+---
+apiVersion: admissionregistration.k8s.io/v1
+kind: MutatingWebhookConfiguration
+metadata:
+  name: kafkabindings.webhook.kafka.sources.knative.dev
+  labels:
+    kafka.eventing.knative.dev/release: devel
+webhooks:
+- admissionReviewVersions: ["v1", "v1beta1"]
+  clientConfig:
+    service:
+      name: kafka-source-webhook
+      namespace: knative-eventing
+  sideEffects: None
+  failurePolicy: Fail
+  name: kafkabindings.webhook.kafka.sources.knative.dev
+---
+apiVersion: admissionregistration.k8s.io/v1
+kind: ValidatingWebhookConfiguration
+metadata:
+  name: validation.webhook.kafka.sources.knative.dev
+  labels:
+    kafka.eventing.knative.dev/release: devel
+webhooks:
+- admissionReviewVersions: ["v1", "v1beta1"]
+  clientConfig:
+    service:
+      name: kafka-source-webhook
+      namespace: knative-eventing
+  sideEffects: None
+  failurePolicy: Fail
+  name: validation.webhook.kafka.sources.knative.dev
+---
+apiVersion: admissionregistration.k8s.io/v1
+kind: ValidatingWebhookConfiguration
+metadata:
+  name: config.webhook.kafka.sources.knative.dev
+  labels:
+    kafka.eventing.knative.dev/release: devel
+webhooks:
+- admissionReviewVersions: ["v1", "v1beta1"]
+  clientConfig:
+    service:
+      name: kafka-source-webhook
+      namespace: knative-eventing
+  sideEffects: None
+  failurePolicy: Fail
+  name: config.webhook.kafka.sources.knative.dev
+  namespaceSelector:
+    matchExpressions:
+    - key: kafka.eventing.knative.dev/release
+      operator: Exists
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: kafka-source-webhook-certs
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
+# The data is populated at install time.
diff --git a/go.mod b/go.mod
index 555f33b6e..bd1441b76 100644
--- a/go.mod
+++ b/go.mod
@@ -28,11 +28,12 @@ require (
 	go.uber.org/zap v1.16.0
 	golang.org/x/crypto v0.0.0-20201016220609-9e8e0b390897 // indirect
 	golang.org/x/sync v0.0.0-20201207232520-09787c993a3a
+	golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e
 	k8s.io/api v0.19.7
 	k8s.io/apimachinery v0.19.7
 	k8s.io/client-go v0.19.7
 	k8s.io/utils v0.0.0-20200729134348-d5654de09c73
-	knative.dev/eventing v0.20.1-0.20210130010931-b20c96b4df51
+	knative.dev/eventing v0.20.1-0.20210201132532-321a8399379c
 	knative.dev/hack v0.0.0-20210120165453-8d623a0af457
 	knative.dev/pkg v0.0.0-20210130001831-ca02ef752ac6
 )
diff --git a/go.sum b/go.sum
index 223936ba4..09800ca0c 100644
--- a/go.sum
+++ b/go.sum
@@ -1246,14 +1246,12 @@ k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6 h1:+WnxoVtG8TMiudHBSEtrVL
 k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6/go.mod h1:UuqjUnNftUyPE5H64/qeyjQoUZhGpeFDVdxjTeEVN2o=
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73 h1:uJmqzgNWG7XyClnU/mLPBWwfKKF1K8Hf8whTseBgJcg=
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.20.1-0.20210130010931-b20c96b4df51 h1:wF7fkEixWzz8wm2lbeI5BVHApPsQEpBEQmhhtt9zh/Q=
-knative.dev/eventing v0.20.1-0.20210130010931-b20c96b4df51/go.mod h1:YzX3ke/NGrBMpLg/LYD1SzHpqN8lpnDysX8Ij0jpbO8=
+knative.dev/eventing v0.20.1-0.20210201132532-321a8399379c h1:i/Uv8syphQh8bEdCdCgFFkgjsaJ3JZ9E4xRe7RTAHTA=
+knative.dev/eventing v0.20.1-0.20210201132532-321a8399379c/go.mod h1:rWfYu3gwM/8Zu+2kInU2iYk9woIDnLAGhzxQCT5eNow=
 knative.dev/hack v0.0.0-20210120165453-8d623a0af457 h1:jEBITgx/lQydGncM0uetpv/ZqawRzb2aSfEaYoMeDjM=
 knative.dev/hack v0.0.0-20210120165453-8d623a0af457/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/pkg v0.0.0-20210125222030-6040b3af4803 h1:L9lY/UztepeiTWys3VyO3Y5wQmYPyYG7eOG9qib6AZY=
 knative.dev/pkg v0.0.0-20210125222030-6040b3af4803/go.mod h1:X4NPrCo8NK3hbDVan9Vm7mf5io3ZoINakAdrpSXVB08=
-knative.dev/pkg v0.0.0-20210127163530-0d31134d5f4e h1:gChYSjSASTy59BGX9R1zaZy7s4tP9EFhAA7WMwQo8nY=
-knative.dev/pkg v0.0.0-20210127163530-0d31134d5f4e/go.mod h1:X4NPrCo8NK3hbDVan9Vm7mf5io3ZoINakAdrpSXVB08=
 knative.dev/pkg v0.0.0-20210130001831-ca02ef752ac6 h1:HIACRvhv/4U7vcFTAakfqYJIlENCSGtTrZOwz/q/A00=
 knative.dev/pkg v0.0.0-20210130001831-ca02ef752ac6/go.mod h1:X4NPrCo8NK3hbDVan9Vm7mf5io3ZoINakAdrpSXVB08=
 knative.dev/reconciler-test v0.0.0-20210127151131-28c3852c3e9b/go.mod h1:+IqX+lV5lQ6AhHY7fomh3sn6K2kW1cTQvC7US5cJB7k=
diff --git a/hack/release.sh b/hack/release.sh
index 0f264fd76..f1d1cc144 100755
--- a/hack/release.sh
+++ b/hack/release.sh
@@ -27,6 +27,7 @@ COMPONENTS=(
   ["channel-consolidated.yaml"]="config/channel/consolidated"
   ["channel-distributed.yaml"]="config/channel/distributed"
   ["source.yaml"]="config/source/single"
+  ["mt-source.yaml"]="config/source/multi"
 )
 readonly COMPONENTS
 
diff --git a/hack/update-codegen.sh b/hack/update-codegen.sh
index 7adbd8bd0..a795c2f17 100755
--- a/hack/update-codegen.sh
+++ b/hack/update-codegen.sh
@@ -36,6 +36,13 @@ ${CODEGEN_PKG}/generate-groups.sh "deepcopy,client,informer,lister" \
 "sources:v1alpha1 sources:v1beta1 bindings:v1alpha1 bindings:v1beta1 messaging:v1alpha1 messaging:v1beta1" \
 --go-header-file ${REPO_ROOT_DIR}/hack/boilerplate.go.txt
 
+# Only deepcopy the Duck types, as they are not real resources.
+chmod +x ${CODEGEN_PKG}/generate-groups.sh
+${CODEGEN_PKG}/generate-groups.sh "deepcopy" \
+  knative.dev/eventing-kafka/pkg/client knative.dev/eventing-kafka/pkg/apis \
+  "duck:v1alpha1" \
+  --go-header-file ${REPO_ROOT_DIR}/hack/boilerplate.go.txt
+
 group "Knative Codegen"
 
 # Knative Injection
diff --git a/pkg/apis/duck/v1alpha1/doc.go b/pkg/apis/duck/v1alpha1/doc.go
new file mode 100644
index 000000000..deae465ed
--- /dev/null
+++ b/pkg/apis/duck/v1alpha1/doc.go
@@ -0,0 +1,19 @@
+/*
+Copyright 2020 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.
+*/
+
+// +k8s:deepcopy-gen=package
+// +groupName=duck.knative.dev
+package v1alpha1
diff --git a/pkg/apis/duck/v1alpha1/placement_types.go b/pkg/apis/duck/v1alpha1/placement_types.go
new file mode 100644
index 000000000..a811b022b
--- /dev/null
+++ b/pkg/apis/duck/v1alpha1/placement_types.go
@@ -0,0 +1,85 @@
+/*
+Copyright 2020 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 v1alpha1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"knative.dev/pkg/apis/duck"
+)
+
+// +genduck
+
+// Placeable is a list of podName and virtual replicas pairs.
+// Each pair represents the assignment of virtual replicas to a pod
+type Placeable struct {
+	Placement []Placement `json:"placements,omitempty"`
+}
+
+// PlaceableType is a skeleton type wrapping Placeable in the manner we expect
+// resource writers defining compatible resources to embed it.  We will
+// typically use this type to deserialize Placeable ObjectReferences and
+// access the Placeable data.  This is not a real resource.
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+type PlaceableType struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Status PlaceableStatus `json:"status"`
+}
+
+type PlaceableStatus struct {
+	Placeable `json:",inline"`
+}
+
+type Placement struct {
+	// PodName is the name of the pod where the resource is placed
+	PodName string `json:"podName,omitempty"`
+
+	// VReplicas is the number of virtual replicas assigned to in the pod
+	VReplicas int32 `json:"vreplicas,omitempty"`
+}
+
+var (
+	// Placeable is an Implementable "duck type".
+	_ duck.Implementable = (*Placeable)(nil)
+)
+
+// GetFullType implements duck.Implementable
+func (*Placeable) GetFullType() duck.Populatable {
+	return &PlaceableType{}
+}
+
+// Populate implements duck.Populatable
+func (t *PlaceableType) Populate() {
+	t.Status.Placement = []Placement{{PodName: "pod-0", VReplicas: int32(1)}}
+}
+
+// GetListType implements apis.Listable
+func (*PlaceableType) GetListType() runtime.Object {
+	return &PlaceableList{}
+}
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// PlaceableList is a list of PlaceableType resources
+type PlaceableList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata"`
+
+	Items []Placeable `json:"items"`
+}
diff --git a/pkg/apis/duck/v1alpha1/register.go b/pkg/apis/duck/v1alpha1/register.go
new file mode 100644
index 000000000..505731628
--- /dev/null
+++ b/pkg/apis/duck/v1alpha1/register.go
@@ -0,0 +1,53 @@
+/*
+Copyright 2020 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 v1alpha1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"knative.dev/pkg/apis/duck"
+)
+
+// SchemeGroupVersion is group version used to register these objects
+var SchemeGroupVersion = schema.GroupVersion{Group: duck.GroupName, Version: "v1alpha1"}
+
+// Kind takes an unqualified kind and returns back a Group qualified GroupKind
+func Kind(kind string) schema.GroupKind {
+	return SchemeGroupVersion.WithKind(kind).GroupKind()
+}
+
+// Resource takes an unqualified resource and returns a Group qualified GroupResource
+func Resource(resource string) schema.GroupResource {
+	return SchemeGroupVersion.WithResource(resource).GroupResource()
+}
+
+var (
+	SchemeBuilder = runtime.NewSchemeBuilder(addKnownTypes)
+	AddToScheme   = SchemeBuilder.AddToScheme
+)
+
+// Adds the list of known types to Scheme.
+func addKnownTypes(scheme *runtime.Scheme) error {
+	scheme.AddKnownTypes(
+		SchemeGroupVersion,
+		&PlaceableType{},
+		(&PlaceableType{}).GetListType(),
+	)
+	metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
+	return nil
+}
diff --git a/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
new file mode 100644
index 000000000..de858a48b
--- /dev/null
+++ b/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
@@ -0,0 +1,139 @@
+// +build !ignore_autogenerated
+
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by deepcopy-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	runtime "k8s.io/apimachinery/pkg/runtime"
+)
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Placeable) DeepCopyInto(out *Placeable) {
+	*out = *in
+	if in.Placement != nil {
+		in, out := &in.Placement, &out.Placement
+		*out = make([]Placement, len(*in))
+		copy(*out, *in)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Placeable.
+func (in *Placeable) DeepCopy() *Placeable {
+	if in == nil {
+		return nil
+	}
+	out := new(Placeable)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *PlaceableList) DeepCopyInto(out *PlaceableList) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ListMeta.DeepCopyInto(&out.ListMeta)
+	if in.Items != nil {
+		in, out := &in.Items, &out.Items
+		*out = make([]Placeable, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PlaceableList.
+func (in *PlaceableList) DeepCopy() *PlaceableList {
+	if in == nil {
+		return nil
+	}
+	out := new(PlaceableList)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *PlaceableList) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *PlaceableStatus) DeepCopyInto(out *PlaceableStatus) {
+	*out = *in
+	in.Placeable.DeepCopyInto(&out.Placeable)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PlaceableStatus.
+func (in *PlaceableStatus) DeepCopy() *PlaceableStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(PlaceableStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *PlaceableType) DeepCopyInto(out *PlaceableType) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	in.Status.DeepCopyInto(&out.Status)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PlaceableType.
+func (in *PlaceableType) DeepCopy() *PlaceableType {
+	if in == nil {
+		return nil
+	}
+	out := new(PlaceableType)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *PlaceableType) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Placement) DeepCopyInto(out *Placement) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Placement.
+func (in *Placement) DeepCopy() *Placement {
+	if in == nil {
+		return nil
+	}
+	out := new(Placement)
+	in.DeepCopyInto(out)
+	return out
+}
diff --git a/pkg/apis/sources/v1alpha1/kafka_conversion.go b/pkg/apis/sources/v1alpha1/kafka_conversion.go
index 65762204c..37787e770 100644
--- a/pkg/apis/sources/v1alpha1/kafka_conversion.go
+++ b/pkg/apis/sources/v1alpha1/kafka_conversion.go
@@ -47,6 +47,7 @@ func (source *KafkaSource) ConvertTo(ctx context.Context, obj apis.Convertible)
 		}
 		source.Status.Status.DeepCopyInto(&sink.Status.Status)
 		sink.Status.Consumers = source.Status.Consumers
+		source.Status.Placeable.DeepCopyInto(&sink.Status.Placeable)
 
 		// Optionals
 		if source.Spec.Sink != nil {
@@ -94,6 +95,7 @@ func (sink *KafkaSource) ConvertFrom(ctx context.Context, obj apis.Convertible)
 
 		source.Status.Status.DeepCopyInto(&sink.Status.Status)
 		sink.Status.Consumers = source.Status.Consumers
+		source.Status.Placeable.DeepCopyInto(&sink.Status.Placeable)
 
 		// Optionals
 		if source.Status.SinkURI != nil {
diff --git a/pkg/apis/sources/v1alpha1/kafka_types.go b/pkg/apis/sources/v1alpha1/kafka_types.go
index 4bdb9f795..3fdbc71f6 100644
--- a/pkg/apis/sources/v1alpha1/kafka_types.go
+++ b/pkg/apis/sources/v1alpha1/kafka_types.go
@@ -22,11 +22,13 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	bindingsv1alpha1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1alpha1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/webhook/resourcesemantics"
+
+	bindingsv1alpha1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1alpha1"
+	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
 )
 
 // +genclient
@@ -134,6 +136,11 @@ type KafkaSourceStatus struct {
 	// +optional
 	// For round-tripping only.
 	Consumers int32 `json:"consumers,omitempty"`
+
+	// Implement Placeable.
+	// For round-tripping only.
+	// +optional
+	duckv1alpha1.Placeable `json:",inline"`
 }
 
 func (*KafkaSource) GetGroupVersionKind() schema.GroupVersionKind {
diff --git a/pkg/apis/sources/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/sources/v1alpha1/zz_generated.deepcopy.go
index e57ddec9e..7c8bf0416 100644
--- a/pkg/apis/sources/v1alpha1/zz_generated.deepcopy.go
+++ b/pkg/apis/sources/v1alpha1/zz_generated.deepcopy.go
@@ -178,6 +178,7 @@ func (in *KafkaSourceSpec) DeepCopy() *KafkaSourceSpec {
 func (in *KafkaSourceStatus) DeepCopyInto(out *KafkaSourceStatus) {
 	*out = *in
 	in.SourceStatus.DeepCopyInto(&out.SourceStatus)
+	in.Placeable.DeepCopyInto(&out.Placeable)
 	return
 }
 
diff --git a/pkg/apis/sources/v1beta1/kafka_lifecycle.go b/pkg/apis/sources/v1beta1/kafka_lifecycle.go
index ddd07ab3d..cab230dcb 100644
--- a/pkg/apis/sources/v1beta1/kafka_lifecycle.go
+++ b/pkg/apis/sources/v1beta1/kafka_lifecycle.go
@@ -17,6 +17,8 @@ limitations under the License.
 package v1beta1
 
 import (
+	"sync"
+
 	appsv1 "k8s.io/api/apps/v1"
 	"knative.dev/eventing/pkg/apis/duck"
 	"knative.dev/pkg/apis"
@@ -37,9 +39,21 @@ const (
 	KafkaConditionKeyType apis.ConditionType = "KeyTypeCorrect"
 )
 
-var KafkaSourceCondSet = apis.NewLivingConditionSet(
-	KafkaConditionSinkProvided,
-	KafkaConditionDeployed)
+var (
+	KafkaSourceCondSet = apis.NewLivingConditionSet(
+		KafkaConditionSinkProvided,
+		KafkaConditionDeployed)
+
+	kafkaCondSetLock = sync.RWMutex{}
+)
+
+// RegisterAlternateKafkaConditionSet register an alternate apis.ConditionSet.
+func RegisterAlternateKafkaConditionSet(conditionSet apis.ConditionSet) {
+	kafkaCondSetLock.Lock()
+	defer kafkaCondSetLock.Unlock()
+
+	KafkaSourceCondSet = conditionSet
+}
 
 // GetConditionSet retrieves the condition set for this resource. Implements the KRShaped interface.
 func (*KafkaSource) GetConditionSet() apis.ConditionSet {
diff --git a/pkg/apis/sources/v1beta1/kafka_lifecycle_mt.go b/pkg/apis/sources/v1beta1/kafka_lifecycle_mt.go
new file mode 100644
index 000000000..3725e0ddd
--- /dev/null
+++ b/pkg/apis/sources/v1beta1/kafka_lifecycle_mt.go
@@ -0,0 +1,39 @@
+/*
+Copyright 2020 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 v1beta1
+
+import (
+	"knative.dev/pkg/apis"
+)
+
+const (
+
+	// KafkaConditionScheduled is True when all KafkaSource consumers has been scheduled
+	KafkaConditionScheduled apis.ConditionType = "Scheduled"
+)
+
+var (
+	KafkaMTSourceCondSet = apis.NewLivingConditionSet(KafkaConditionSinkProvided, KafkaConditionScheduled)
+)
+
+func (s *KafkaSourceStatus) MarkScheduled() {
+	KafkaSourceCondSet.Manage(s).MarkTrue(KafkaConditionScheduled)
+}
+
+func (s *KafkaSourceStatus) MarkNotScheduled(reason, messageFormat string, messageA ...interface{}) {
+	KafkaSourceCondSet.Manage(s).MarkFalse(KafkaConditionScheduled, reason, messageFormat, messageA...)
+}
diff --git a/pkg/apis/sources/v1beta1/kafka_scheduling.go b/pkg/apis/sources/v1beta1/kafka_scheduling.go
new file mode 100644
index 000000000..293ecda6f
--- /dev/null
+++ b/pkg/apis/sources/v1beta1/kafka_scheduling.go
@@ -0,0 +1,43 @@
+/*
+Copyright 2020 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 v1beta1
+
+import (
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+)
+
+func (k *KafkaSource) GetKey() types.NamespacedName {
+	return types.NamespacedName{
+		Namespace: k.Namespace,
+		Name:      k.Name,
+	}
+}
+
+func (k *KafkaSource) GetVReplicas() int32 {
+	if k.Spec.Consumers == nil {
+		return 1
+	}
+	return *k.Spec.Consumers
+}
+
+func (k *KafkaSource) GetPlacements() []v1alpha1.Placement {
+	if k.Status.Placeable.Placement == nil {
+		return nil
+	}
+	return k.Status.Placeable.Placement
+}
diff --git a/pkg/apis/sources/v1beta1/kafka_types.go b/pkg/apis/sources/v1beta1/kafka_types.go
index 0af251ff4..7d4329d50 100644
--- a/pkg/apis/sources/v1beta1/kafka_types.go
+++ b/pkg/apis/sources/v1beta1/kafka_types.go
@@ -19,6 +19,8 @@ package v1beta1
 import (
 	"fmt"
 
+	"knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -30,6 +32,7 @@ import (
 )
 
 // +genclient
+// +genclient:method=GetScale,verb=get,subresource=scale,result=k8s.io/api/autoscaling/v1.Scale
 // +genreconciler
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 // KafkaSource is the Schema for the kafkasources API.
@@ -105,6 +108,10 @@ type KafkaSourceStatus struct {
 	// Total number of consumers actually running in the consumer group.
 	// +optional
 	Consumers int32 `json:"consumers,omitempty"`
+
+	// Implement Placeable.
+	// +optional
+	v1alpha1.Placeable `json:",inline"`
 }
 
 func (*KafkaSource) GetGroupVersionKind() schema.GroupVersionKind {
diff --git a/pkg/apis/sources/v1beta1/zz_generated.deepcopy.go b/pkg/apis/sources/v1beta1/zz_generated.deepcopy.go
index 025c8e7de..f494920ef 100644
--- a/pkg/apis/sources/v1beta1/zz_generated.deepcopy.go
+++ b/pkg/apis/sources/v1beta1/zz_generated.deepcopy.go
@@ -117,6 +117,7 @@ func (in *KafkaSourceSpec) DeepCopy() *KafkaSourceSpec {
 func (in *KafkaSourceStatus) DeepCopyInto(out *KafkaSourceStatus) {
 	*out = *in
 	in.SourceStatus.DeepCopyInto(&out.SourceStatus)
+	in.Placeable.DeepCopyInto(&out.Placeable)
 	return
 }
 
diff --git a/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_kafkasource.go b/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_kafkasource.go
index 978466a6c..a35a59c17 100644
--- a/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_kafkasource.go
+++ b/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_kafkasource.go
@@ -21,6 +21,7 @@ package fake
 import (
 	"context"
 
+	autoscalingv1 "k8s.io/api/autoscaling/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	labels "k8s.io/apimachinery/pkg/labels"
 	schema "k8s.io/apimachinery/pkg/runtime/schema"
@@ -140,3 +141,14 @@ func (c *FakeKafkaSources) Patch(ctx context.Context, name string, pt types.Patc
 	}
 	return obj.(*v1beta1.KafkaSource), err
 }
+
+// GetScale takes name of the kafkaSource, and returns the corresponding scale object, and an error if there is any.
+func (c *FakeKafkaSources) GetScale(ctx context.Context, kafkaSourceName string, options v1.GetOptions) (result *autoscalingv1.Scale, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetSubresourceAction(kafkasourcesResource, c.ns, "scale", kafkaSourceName), &autoscalingv1.Scale{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*autoscalingv1.Scale), err
+}
diff --git a/pkg/client/clientset/versioned/typed/sources/v1beta1/kafkasource.go b/pkg/client/clientset/versioned/typed/sources/v1beta1/kafkasource.go
index 94c619d6c..d6d330585 100644
--- a/pkg/client/clientset/versioned/typed/sources/v1beta1/kafkasource.go
+++ b/pkg/client/clientset/versioned/typed/sources/v1beta1/kafkasource.go
@@ -22,6 +22,7 @@ import (
 	"context"
 	"time"
 
+	autoscalingv1 "k8s.io/api/autoscaling/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	types "k8s.io/apimachinery/pkg/types"
 	watch "k8s.io/apimachinery/pkg/watch"
@@ -47,6 +48,8 @@ type KafkaSourceInterface interface {
 	List(ctx context.Context, opts v1.ListOptions) (*v1beta1.KafkaSourceList, error)
 	Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error)
 	Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1beta1.KafkaSource, err error)
+	GetScale(ctx context.Context, kafkaSourceName string, options v1.GetOptions) (*autoscalingv1.Scale, error)
+
 	KafkaSourceExpansion
 }
 
@@ -193,3 +196,17 @@ func (c *kafkaSources) Patch(ctx context.Context, name string, pt types.PatchTyp
 		Into(result)
 	return
 }
+
+// GetScale takes name of the kafkaSource, and returns the corresponding autoscalingv1.Scale object, and an error if there is any.
+func (c *kafkaSources) GetScale(ctx context.Context, kafkaSourceName string, options v1.GetOptions) (result *autoscalingv1.Scale, err error) {
+	result = &autoscalingv1.Scale{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("kafkasources").
+		Name(kafkaSourceName).
+		SubResource("scale").
+		VersionedParams(&options, scheme.ParameterCodec).
+		Do(ctx).
+		Into(result)
+	return
+}
diff --git a/pkg/common/consumer/consumer_handler.go b/pkg/common/consumer/consumer_handler.go
index 7722ec0a3..b27ccecdb 100644
--- a/pkg/common/consumer/consumer_handler.go
+++ b/pkg/common/consumer/consumer_handler.go
@@ -71,6 +71,7 @@ func (consumer *SaramaConsumerHandler) ConsumeClaim(session sarama.ConsumerGroup
 	// The `ConsumeClaim` itself is called within a goroutine, see:
 	// https://github.com/Shopify/sarama/blob/master/consumer_group.go#L27-L29
 	for message := range claim.Messages() {
+
 		if ce := consumer.logger.Desugar().Check(zap.DebugLevel, "debugging"); ce != nil {
 			consumer.logger.Debugw("Message claimed", zap.String("topic", message.Topic), zap.Binary("value", message.Value))
 		}
diff --git a/pkg/common/scheduler/doc.go b/pkg/common/scheduler/doc.go
new file mode 100644
index 000000000..57ebf7e89
--- /dev/null
+++ b/pkg/common/scheduler/doc.go
@@ -0,0 +1,18 @@
+/*
+Copyright 2020 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.
+*/
+
+// The scheduler is responsible for placing virtual pod (VPod) replicas within real pods.
+package scheduler
diff --git a/pkg/common/scheduler/placement.go b/pkg/common/scheduler/placement.go
new file mode 100644
index 000000000..ec47f8d0c
--- /dev/null
+++ b/pkg/common/scheduler/placement.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2020 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 scheduler
+
+import (
+	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+)
+
+// GetTotalVReplicas returns the total number of placed virtual replicas
+func GetTotalVReplicas(placements []duckv1alpha1.Placement) int32 {
+	r := int32(0)
+	for _, p := range placements {
+		r += p.VReplicas
+	}
+	return r
+}
+
+// GetPlacementForPod returns the placement corresponding to podName
+func GetPlacementForPod(placements []duckv1alpha1.Placement, podName string) *duckv1alpha1.Placement {
+	for i := 0; i < len(placements); i++ {
+		if placements[i].PodName == podName {
+			return &placements[i]
+		}
+	}
+	return nil
+}
diff --git a/pkg/common/scheduler/scheduler.go b/pkg/common/scheduler/scheduler.go
new file mode 100644
index 000000000..eb47174d9
--- /dev/null
+++ b/pkg/common/scheduler/scheduler.go
@@ -0,0 +1,53 @@
+/*
+Copyright 2020 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 scheduler
+
+import (
+	"errors"
+
+	"k8s.io/apimachinery/pkg/types"
+
+	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+)
+
+var (
+	ErrNoReplicas        = errors.New("scheduling failed (no pod replicas available)")
+	ErrNotEnoughReplicas = errors.New("scheduling partially failed (not enough pod replicas)")
+)
+
+// VPodLister is the function signature for returning a list of VPods
+type VPodLister func() ([]VPod, error)
+
+// Scheduler is responsible for placing VPods into real Kubernetes pods
+type Scheduler interface {
+	// Schedule computes the new set of placements for vpod.
+	Schedule(vpod VPod) ([]duckv1alpha1.Placement, error)
+}
+
+// VPod represents virtual replicas placed into real Kubernetes pods
+// The scheduler is responsible for placing VPods
+type VPod interface {
+	// GetKey returns the VPod key (namespace/name).
+	GetKey() types.NamespacedName
+
+	// GetVReplicas returns the number of expected virtual replicas
+	GetVReplicas() int32
+
+	// GetPlacements returns the current list of placements
+	// Do not mutate!
+	GetPlacements() []duckv1alpha1.Placement
+}
diff --git a/pkg/common/scheduler/statefulset/autoscaler.go b/pkg/common/scheduler/statefulset/autoscaler.go
new file mode 100644
index 000000000..597e9356c
--- /dev/null
+++ b/pkg/common/scheduler/statefulset/autoscaler.go
@@ -0,0 +1,121 @@
+/*
+Copyright 2020 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 statefulset
+
+import (
+	"context"
+	"math"
+	"time"
+
+	"go.uber.org/zap"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func (s *StatefulSetScheduler) autoscale(ctx context.Context) {
+	// Regularly compute the average number of replicas per pods.
+	// When the average goes above a certain ratio, scale up
+	// otherwise scale down.
+
+	for {
+		func() {
+			s.lock.Lock()
+			defer s.lock.Unlock()
+
+			// Refresh state as the scheduler does not receive deletion notifications.
+			snapshot, err := s.snapshot(s.logger)
+			if err != nil {
+				s.logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
+				return
+			}
+
+			s.logger.Infow("checking adapter capacity",
+				zap.Int32("free", s.freeCapacity(snapshot)),
+				zap.Int32("used", s.usedCapacity(snapshot)),
+				zap.Int32("pending", s.pendingVReplicas()),
+				zap.Int32("replicas", s.replicas))
+
+			var ratio float64
+			if s.replicas == 0 {
+				if s.pendingVReplicas() > 0 {
+					ratio = 1.0
+				} else {
+					ratio = 0.5
+				}
+			} else {
+				ratio = s.avgUsedCapacityPerPod(snapshot) / float64(s.capacity)
+			}
+
+			if ratio > 0.7 || ratio < 0.3 {
+				// Scale up when capacity is above 80% (TODO: configurable)
+				// Scale down when capacity is below 30% (TODO: configurable)
+				s.logger.Infow("autoscaling statefulset", zap.Int("avg used capacity per pod", int(ratio*100.0)))
+
+				scale, err := s.statefulSetClient.GetScale(ctx, s.statefulSetName, metav1.GetOptions{})
+				if err != nil {
+					// skip a beat
+					s.logger.Infow("failed to get scale subresource", zap.Error(err))
+					return
+				}
+
+				// Desired ratio is 0.5 (TODO: configurable)
+				scale.Spec.Replicas = 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))
+
+				_, err = s.statefulSetClient.UpdateScale(ctx, s.statefulSetName, scale, metav1.UpdateOptions{})
+				if err != nil {
+					s.logger.Errorw("updating scale subresource failed", zap.Error(err))
+				}
+			}
+		}()
+
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(10 * time.Second): // TODO: configurable refresh period
+		}
+	}
+}
+
+// freeCapacity returns the free capacity across all pods
+func (s *StatefulSetScheduler) freeCapacity(snapshot *Snapshot) int32 {
+	t := int32(0)
+	for i := int32(0); i < s.replicas; i++ {
+		t += snapshot.free[podNameFromOrdinal(s.statefulSetName, i)]
+	}
+	return t
+}
+
+// usedCapacity returns the used capacity across all pods
+func (s *StatefulSetScheduler) usedCapacity(snapshot *Snapshot) int32 {
+	return s.capacity*s.replicas - s.freeCapacity(snapshot)
+}
+
+// pendingReplicas returns the total number of virtual replicas
+// that haven't been scheduled yet
+func (s *StatefulSetScheduler) pendingVReplicas() int32 {
+	t := int32(0)
+	for _, v := range s.pending {
+		t += v
+	}
+	return t
+}
+
+// avgUsedCapacityPerPod returns the average used capacity per replica
+func (s *StatefulSetScheduler) avgUsedCapacityPerPod(snapshot *Snapshot) float64 {
+	return float64(s.usedCapacity(snapshot)) / float64(s.replicas)
+}
diff --git a/pkg/common/scheduler/statefulset/helpers.go b/pkg/common/scheduler/statefulset/helpers.go
new file mode 100644
index 000000000..2d545f1a4
--- /dev/null
+++ b/pkg/common/scheduler/statefulset/helpers.go
@@ -0,0 +1,34 @@
+/*
+Copyright 2020 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 statefulset
+
+import (
+	"math"
+	"strconv"
+	"strings"
+)
+
+func podNameFromOrdinal(name string, ordinal int32) string {
+	return name + "-" + strconv.Itoa(int(ordinal))
+}
+
+func ordinalFromPodName(podName string) int32 {
+	ordinal, err := strconv.ParseInt(podName[strings.LastIndex(podName, "-")+1:], 10, 32)
+	if err != nil {
+		return math.MaxInt32
+	}
+	return int32(ordinal)
+}
diff --git a/pkg/common/scheduler/statefulset/scheduler.go b/pkg/common/scheduler/statefulset/scheduler.go
new file mode 100644
index 000000000..8796bad45
--- /dev/null
+++ b/pkg/common/scheduler/statefulset/scheduler.go
@@ -0,0 +1,277 @@
+/*
+Copyright 2020 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 statefulset
+
+import (
+	"context"
+	"sync"
+
+	"go.uber.org/zap"
+	appsv1 "k8s.io/api/apps/v1"
+	clientappsv1 "k8s.io/client-go/kubernetes/typed/apps/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/utils/integer"
+
+	"k8s.io/apimachinery/pkg/types"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	statefulsetinformer "knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+
+	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+	"knative.dev/eventing-kafka/pkg/common/scheduler"
+)
+
+// StatefulSetScheduler is a scheduler placing VPod into statefulset-managed pods
+type StatefulSetScheduler struct {
+	logger            *zap.SugaredLogger
+	statefulSetName   string
+	statefulSetClient clientappsv1.StatefulSetInterface
+	vpodLister        scheduler.VPodLister
+	lock              sync.Locker
+
+	// replicas is the (cached) number of statefulset replicas
+	replicas int32
+
+	// capacity is the total number of virtual replicas available per pod.
+	capacity int32
+
+	// pending tracks the number of virtual replicas that haven't been scheduled yet.
+	// Used by the autoscaler
+	pending map[types.NamespacedName]int32
+
+	// enq is used to trigger the reconciliation of the named object
+	enq func(types.NamespacedName)
+}
+
+func NewStatefulSetScheduler(ctx context.Context, namespace, name string, lister scheduler.VPodLister, enq func(types.NamespacedName)) scheduler.Scheduler {
+	scheduler := &StatefulSetScheduler{
+		logger:            logging.FromContext(ctx),
+		statefulSetName:   name,
+		statefulSetClient: kubeclient.Get(ctx).AppsV1().StatefulSets(namespace),
+		vpodLister:        lister,
+		capacity:          10, // TODO: from annotation on statefulset, configmap?
+		enq:               enq,
+		pending:           make(map[types.NamespacedName]int32),
+		lock:              new(sync.Mutex),
+	}
+
+	// Monitor our statefulset
+	statefulsetInformer := statefulsetinformer.Get(ctx)
+	statefulsetInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: controller.FilterWithNameAndNamespace(namespace, name),
+		Handler:    controller.HandleAll(scheduler.updateStatefulset),
+	})
+
+	// Start the autoscaler. Eventually we may want to rely on HPA + custom metrics
+	// in particular to handle fluctuation
+	go scheduler.autoscale(ctx)
+
+	return scheduler
+}
+
+func (s *StatefulSetScheduler) Schedule(vpod scheduler.VPod) ([]duckv1alpha1.Placement, error) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	logger := s.logger.With("key", zap.Any("key", vpod.GetKey()))
+	logger.Info("scheduling")
+
+	// Get the current placement state
+	// Quite an expensive operation but safe and simple.
+	snapshot, err := s.snapshot(logger)
+	if err != nil {
+		logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
+		return nil, err
+	}
+
+	if s.replicas == 0 {
+		logger.Info("scheduling failed (no replicas available)")
+		s.pending[vpod.GetKey()] = vpod.GetVReplicas()
+		return make([]duckv1alpha1.Placement, 0), scheduler.ErrNoReplicas
+	}
+
+	// Remove placements from scaled-down pods
+	placements := s.reschedule(vpod.GetPlacements())
+
+	// The scheduler
+	// - allocates as many vreplicas as possible to the same pod(s) (for placed vpod)
+	// - allocates remaining vreplicas to new pods
+
+	// Exact number of vreplicas => do nothing
+	tr := scheduler.GetTotalVReplicas(placements)
+	if tr == vpod.GetVReplicas() {
+		logger.Info("scheduling succeeded (already scheduled)")
+		delete(s.pending, vpod.GetKey())
+
+		// Fully placed. Nothing to do
+		return placements, nil
+	}
+
+	// Need less => scale down
+	if tr > vpod.GetVReplicas() {
+		logger.Infow("scaling down", zap.Int32("vreplicas", tr), zap.Int32("new vreplicas", vpod.GetVReplicas()))
+		placements := s.removeReplicas(tr-vpod.GetVReplicas(), placements)
+		return placements, nil
+	}
+
+	// Need more => scale up
+	logger.Infow("scaling up", zap.Int32("vreplicas", tr), zap.Int32("new vreplicas", vpod.GetVReplicas()))
+	placements, left := s.addReplicas(snapshot, vpod.GetVReplicas()-tr, placements)
+	if left > 0 {
+		// Give time for the autoscaler to do its job
+		logger.Info("scheduling failed (not enough pod replicas)", zap.Any("placement", placements))
+
+		s.pending[vpod.GetKey()] = left
+		return placements, scheduler.ErrNotEnoughReplicas
+	}
+
+	logger.Infow("scheduling successful", zap.Any("placement", placements))
+	delete(s.pending, vpod.GetKey())
+	return placements, nil
+}
+
+func (s *StatefulSetScheduler) reschedule(placements []duckv1alpha1.Placement) []duckv1alpha1.Placement {
+	if s.needRescheduling(placements) {
+		newPlacements := make([]duckv1alpha1.Placement, 0, len(placements))
+		for i := 0; i < len(placements); i++ {
+			if ordinalFromPodName(placements[i].PodName) >= s.replicas {
+				newPlacements = append(newPlacements, placements[i])
+			}
+		}
+		return newPlacements
+	}
+	return placements
+}
+
+func (s *StatefulSetScheduler) needRescheduling(placements []duckv1alpha1.Placement) bool {
+	for i := 0; i < len(placements); i++ {
+		if ordinalFromPodName(placements[i].PodName) >= s.replicas {
+			return true
+		}
+	}
+	return false
+}
+
+func (s *StatefulSetScheduler) removeReplicas(diff int32, placements []duckv1alpha1.Placement) []duckv1alpha1.Placement {
+	newPlacements := make([]duckv1alpha1.Placement, 0, len(placements))
+	for i := 0; i < len(placements); i++ {
+		if diff >= placements[i].VReplicas {
+			// remove the entire placement
+			diff -= placements[i].VReplicas
+		} else {
+			newPlacements = append(newPlacements, duckv1alpha1.Placement{
+				PodName:   placements[i].PodName,
+				VReplicas: placements[i].VReplicas - diff,
+			})
+			diff = 0
+		}
+	}
+	return newPlacements
+}
+
+func (s *StatefulSetScheduler) addReplicas(snapshot *Snapshot, diff int32, placements []duckv1alpha1.Placement) ([]duckv1alpha1.Placement, int32) {
+	// Pod affinity algorithm: prefer adding replicas to existing pods before considering other replicas
+	// In the future, we might want to spread replicas across pods in different regions.
+	newPlacements := make([]duckv1alpha1.Placement, 0, len(placements))
+
+	// Add to existing
+	for i := 0; i < len(placements); i++ {
+		podName := placements[i].PodName
+
+		// Is there space in PodName?
+		if diff >= 0 && snapshot.free[podName] > 0 {
+			allocation := integer.Int32Min(snapshot.free[podName], diff)
+			newPlacements = append(newPlacements, duckv1alpha1.Placement{
+				PodName:   podName,
+				VReplicas: placements[i].VReplicas + allocation,
+			})
+
+			diff -= allocation
+			snapshot.free[podName] -= allocation
+		} else {
+			newPlacements = append(newPlacements, placements[i])
+		}
+	}
+
+	if diff > 0 {
+		// Needs to allocate replicas to additional pods
+		for i := int32(0); i < s.replicas; i++ {
+			podName := podNameFromOrdinal(s.statefulSetName, i)
+
+			if snapshot.free[podName] > 0 {
+				allocation := integer.Int32Min(snapshot.free[podName], diff)
+				newPlacements = append(newPlacements, duckv1alpha1.Placement{
+					PodName:   podName,
+					VReplicas: allocation,
+				})
+
+				diff -= allocation
+				snapshot.free[podName] -= allocation
+			}
+
+			if diff == 0 {
+				break
+			}
+		}
+	}
+
+	return newPlacements, diff
+}
+
+// --- StatefulSet monitoring helpers
+
+func (s *StatefulSetScheduler) updateStatefulset(obj interface{}) {
+	statefulset, ok := obj.(*appsv1.StatefulSet)
+	if !ok {
+		s.logger.Fatalw("expected a Statefulset object", zap.Any("object", obj))
+	}
+
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	if s.replicas < statefulset.Status.Replicas {
+		// Less replicas: evict VPods scheduled on those replicas.
+		s.evict()
+
+	}
+	// else if s.replicas > statefulset.Status.Replicas {
+	//   noop: controller will retry to schedule pending vreplicas
+	// }
+
+	s.replicas = statefulset.Status.Replicas
+
+	s.logger.Infow("updated statefulset replicas", zap.Int32("replicas", s.replicas))
+}
+
+func (s *StatefulSetScheduler) evict() {
+	vpods, err := s.vpodLister()
+	if err != nil {
+		s.logger.Fatalw("failed to list schedulable objects", zap.Error(err))
+		return
+	}
+
+	for _, vpod := range vpods {
+		ps := vpod.GetPlacements()
+		for i := 0; i < len(ps); i++ {
+			if ordinalFromPodName(ps[i].PodName) >= s.replicas {
+				s.enq(vpod.GetKey())
+				break
+			}
+		}
+	}
+}
diff --git a/pkg/common/scheduler/statefulset/snapshot.go b/pkg/common/scheduler/statefulset/snapshot.go
new file mode 100644
index 000000000..a2891c1e3
--- /dev/null
+++ b/pkg/common/scheduler/statefulset/snapshot.go
@@ -0,0 +1,55 @@
+/*
+Copyright 2020 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 statefulset
+
+import "go.uber.org/zap"
+
+// Snapshot provides some information about the current scheduling of all vpods
+type Snapshot struct {
+	// free tracks the free capacity of each pod.
+	free map[string]int32
+}
+
+func (s *StatefulSetScheduler) snapshot(logger *zap.SugaredLogger) (*Snapshot, error) {
+	vpods, err := s.vpodLister()
+	if err != nil {
+		return nil, err
+	}
+
+	free := make(map[string]int32)
+	for i := int32(0); i < s.replicas; i++ {
+		podName := podNameFromOrdinal(s.statefulSetName, i)
+		free[podName] = s.capacity
+	}
+
+	for _, vpod := range vpods {
+		ps := vpod.GetPlacements()
+
+		for i := 0; i < len(ps); i++ {
+			podName := ps[i].PodName
+			vreplicas := ps[i].VReplicas
+
+			free[podName] -= vreplicas
+
+			if free[podName] < 0 {
+				// Over committed. The autoscaler will fix it.
+				logger.Infow("pod is overcommitted", zap.String("podName", podName))
+			}
+		}
+	}
+	return &Snapshot{free: free}, nil
+}
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index 84500f7ad..1450c0ef5 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -17,33 +17,36 @@ limitations under the License.
 package kafka
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"net/http"
 	"strings"
 
-	"knative.dev/eventing-kafka/pkg/source"
+	"golang.org/x/time/rate"
 
+	"github.com/Shopify/sarama"
 	"go.opencensus.io/trace"
-	"knative.dev/eventing/pkg/adapter/v2"
-	"knative.dev/eventing/pkg/kncloudevents"
+	"go.uber.org/zap"
+
+	"knative.dev/pkg/logging"
 	pkgsource "knative.dev/pkg/source"
 
-	"context"
+	"knative.dev/eventing/pkg/adapter/v2"
+	"knative.dev/eventing/pkg/kncloudevents"
 
-	"github.com/Shopify/sarama"
-	"go.uber.org/zap"
 	"knative.dev/eventing-kafka/pkg/common/consumer"
-	"knative.dev/pkg/logging"
+	kafkasource "knative.dev/eventing-kafka/pkg/source"
 )
 
 const (
 	resourceGroup = "kafkasources.sources.knative.dev"
 )
 
-type adapterConfig struct {
+type AdapterConfig struct {
 	adapter.EnvConfig
+	kafkasource.KafkaEnvConfig
 
 	Topics        []string `envconfig:"KAFKA_TOPICS" required:"true"`
 	ConsumerGroup string   `envconfig:"KAFKA_CONSUMER_GROUP" required:"true"`
@@ -52,15 +55,16 @@ type adapterConfig struct {
 }
 
 func NewEnvConfig() adapter.EnvConfigAccessor {
-	return &adapterConfig{}
+	return &AdapterConfig{}
 }
 
 type Adapter struct {
-	config            *adapterConfig
+	config            *AdapterConfig
 	httpMessageSender *kncloudevents.HTTPMessageSender
 	reporter          pkgsource.StatsReporter
 	logger            *zap.SugaredLogger
 	keyTypeMapper     func([]byte) interface{}
+	rateLimiter       *rate.Limiter
 }
 
 var _ adapter.MessageAdapter = (*Adapter)(nil)
@@ -68,7 +72,7 @@ var _ adapter.MessageAdapterConstructor = NewAdapter
 
 func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, httpMessageSender *kncloudevents.HTTPMessageSender, reporter pkgsource.StatsReporter) adapter.MessageAdapter {
 	logger := logging.FromContext(ctx)
-	config := processed.(*adapterConfig)
+	config := processed.(*AdapterConfig)
 
 	return &Adapter{
 		config:            config,
@@ -93,7 +97,7 @@ func (a *Adapter) start(stopCh <-chan struct{}) error {
 	)
 
 	// init consumer group
-	addrs, config, err := source.NewConfig(context.Background())
+	addrs, config, err := kafkasource.NewConfigWithEnv(context.Background(), &a.config.KafkaEnvConfig)
 	if err != nil {
 		return fmt.Errorf("failed to create the config: %w", err)
 	}
@@ -118,6 +122,10 @@ func (a *Adapter) start(stopCh <-chan struct{}) error {
 }
 
 func (a *Adapter) Handle(ctx context.Context, msg *sarama.ConsumerMessage) (bool, error) {
+	if a.rateLimiter != nil {
+		a.rateLimiter.Wait(ctx)
+	}
+
 	ctx, span := trace.StartSpan(ctx, "kafka-source")
 	defer span.End()
 
@@ -158,3 +166,8 @@ func (a *Adapter) Handle(ctx context.Context, msg *sarama.ConsumerMessage) (bool
 	_ = a.reporter.ReportEventCount(reportArgs, res.StatusCode)
 	return true, nil
 }
+
+// SetRateLimiter sets the global consumer rate limiter
+func (a *Adapter) SetRateLimits(r rate.Limit, b int) {
+	a.rateLimiter = rate.NewLimiter(r, b)
+}
diff --git a/pkg/source/client.go b/pkg/source/client.go
index 1a52adf1b..11d35da57 100644
--- a/pkg/source/client.go
+++ b/pkg/source/client.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/Shopify/sarama"
 	"github.com/kelseyhightower/envconfig"
+
 	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
 	"knative.dev/eventing-kafka/pkg/common/client"
 )
@@ -45,18 +46,23 @@ type AdapterNet struct {
 	TLS  AdapterTLS
 }
 
-type envConfig struct {
+type KafkaEnvConfig struct {
 	BootstrapServers []string `envconfig:"KAFKA_BOOTSTRAP_SERVERS" required:"true"`
 	Net              AdapterNet
 }
 
 // NewConfig extracts the Kafka configuration from the environment.
 func NewConfig(ctx context.Context) ([]string, *sarama.Config, error) {
-	var env envConfig
+	var env KafkaEnvConfig
 	if err := envconfig.Process("", &env); err != nil {
 		return nil, nil, err
 	}
 
+	return NewConfigWithEnv(ctx, &env)
+}
+
+// NewConfig extracts the Kafka configuration from the environment.
+func NewConfigWithEnv(ctx context.Context, env *KafkaEnvConfig) ([]string, *sarama.Config, error) {
 	kafkaAuthConfig := &client.KafkaAuthConfig{}
 
 	if env.Net.TLS.Enable {
diff --git a/pkg/source/mtadapter/adapter.go b/pkg/source/mtadapter/adapter.go
new file mode 100644
index 000000000..4847a78b0
--- /dev/null
+++ b/pkg/source/mtadapter/adapter.go
@@ -0,0 +1,171 @@
+/*
+Copyright 2020 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 mtadapter
+
+import (
+	"context"
+	"sync"
+
+	"golang.org/x/time/rate"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"go.uber.org/zap"
+
+	"knative.dev/pkg/logging"
+	pkgsource "knative.dev/pkg/source"
+
+	"knative.dev/eventing/pkg/adapter/v2"
+	"knative.dev/eventing/pkg/kncloudevents"
+
+	"knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/eventing-kafka/pkg/common/scheduler"
+	"knative.dev/eventing-kafka/pkg/source"
+	stadapter "knative.dev/eventing-kafka/pkg/source/adapter"
+)
+
+type AdapterConfig struct {
+	adapter.EnvConfig
+
+	PodName string `envconfig:"POD_NAME" required:"true"`
+}
+
+func NewEnvConfig() adapter.EnvConfigAccessor {
+	return new(AdapterConfig)
+}
+
+type Adapter struct {
+	config      *AdapterConfig
+	logger      *zap.SugaredLogger
+	client      cloudevents.Client
+	adapterCtor adapter.MessageAdapterConstructor
+
+	sourcesMu sync.RWMutex
+	sources   map[string]context.CancelFunc
+}
+
+var _ adapter.Adapter = (*Adapter)(nil)
+
+func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ceClient cloudevents.Client) adapter.Adapter {
+	return newAdapter(ctx, processed, ceClient, stadapter.NewAdapter)
+}
+
+func newAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ceClient cloudevents.Client, adapterCtor adapter.MessageAdapterConstructor) adapter.Adapter {
+	logger := logging.FromContext(ctx)
+	config := processed.(*AdapterConfig)
+
+	return &Adapter{
+		client:      ceClient,
+		config:      config,
+		logger:      logger,
+		adapterCtor: adapterCtor,
+		sourcesMu:   sync.RWMutex{},
+		sources:     make(map[string]context.CancelFunc),
+	}
+}
+
+func (a *Adapter) Start(ctx context.Context) error {
+	<-ctx.Done()
+	a.logger.Info("Shutting down...")
+	return nil
+}
+
+// Implements MTAdapter
+
+func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
+	key := obj.Namespace + "/" + obj.Name
+
+	a.sourcesMu.RLock()
+	cancel, ok := a.sources[key]
+	a.sourcesMu.RUnlock()
+
+	if ok {
+		// TODO: do not stop if the only thing that changes is the number of vreplicas
+		a.logger.Info("stopping adapter", zap.String("key", key))
+		cancel()
+	}
+
+	placement := scheduler.GetPlacementForPod(obj.GetPlacements(), a.config.PodName)
+	if placement == nil || placement.VReplicas == 0 {
+		// this pod does not handle this source. Skipping
+		a.logger.Infow("no replicas assigned to this pod. skipping", zap.String("key", key))
+		return
+	}
+
+	config := stadapter.AdapterConfig{
+		EnvConfig: adapter.EnvConfig{
+			Component: "kafkasource",
+		},
+		KafkaEnvConfig: source.KafkaEnvConfig{
+			BootstrapServers: obj.Spec.BootstrapServers,
+		},
+		Topics:        obj.Spec.Topics,
+		ConsumerGroup: obj.Spec.ConsumerGroup,
+		Name:          obj.Name,
+	}
+
+	if val, ok := obj.GetLabels()[v1beta1.KafkaKeyTypeLabel]; ok {
+		config.KeyType = val
+	}
+
+	reporter, err := pkgsource.NewStatsReporter()
+	if err != nil {
+		a.logger.Error("error building statsreporter", zap.Error(err))
+	}
+
+	httpBindingsSender, err := kncloudevents.NewHTTPMessageSenderWithTarget(obj.Status.SinkURI.String())
+	if err != nil {
+		a.logger.Fatal("error building cloud event client", zap.Error(err))
+	}
+
+	adapter := a.adapterCtor(ctx, &config, httpBindingsSender, reporter)
+
+	if a, ok := adapter.(*stadapter.Adapter); ok {
+		// TODO: configurable
+		a.SetRateLimits(rate.Limit(10.0*placement.VReplicas), 20*int(placement.VReplicas))
+	}
+
+	ctx, cancelFn := context.WithCancel(ctx)
+	go func(ctx context.Context) {
+		err := adapter.Start(ctx)
+		if err != nil {
+			a.logger.Error("adapter failed to start", zap.Error(err))
+		}
+
+	}(ctx)
+
+	a.sourcesMu.Lock()
+	a.sources[key] = cancelFn
+	a.sourcesMu.Unlock()
+}
+
+func (a *Adapter) Remove(ctx context.Context, obj *v1beta1.KafkaSource) {
+	key := obj.Namespace + "/" + obj.Name
+
+	a.sourcesMu.RLock()
+	cancel, ok := a.sources[key]
+	a.sourcesMu.RUnlock()
+
+	if !ok {
+		return
+	}
+
+	cancel()
+
+	a.sourcesMu.Lock()
+	delete(a.sources, key)
+	a.sourcesMu.Unlock()
+}
diff --git a/pkg/source/mtadapter/controller.go b/pkg/source/mtadapter/controller.go
new file mode 100644
index 000000000..7801b9a01
--- /dev/null
+++ b/pkg/source/mtadapter/controller.go
@@ -0,0 +1,64 @@
+/*
+Copyright 2020 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 mtadapter
+
+import (
+	"context"
+
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+
+	"knative.dev/eventing/pkg/adapter/v2"
+
+	"knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	kakfasourceinformer "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource"
+	kafkasourcereconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
+)
+
+// MTAdapter is the interface the multi-tenant KafkaSource adapter must implement
+type MTAdapter interface {
+	// Update is called when the source is ready and when the specification and/or status has changed.
+	Update(ctx context.Context, source *v1beta1.KafkaSource)
+
+	// Remove is called when the source has been deleted.
+	Remove(ctx context.Context, source *v1beta1.KafkaSource)
+}
+
+// NewController initializes the controller and
+// registers event handlers to enqueue events.
+func NewController(ctx context.Context, adapter adapter.Adapter) *controller.Impl {
+	mtadapter, ok := adapter.(MTAdapter)
+	if !ok {
+		logging.FromContext(ctx).Fatal("Multi-tenant adapters must implement the MTAdapter interface")
+	}
+
+	r := &Reconciler{
+		mtadapter: mtadapter,
+	}
+
+	impl := kafkasourcereconciler.NewImpl(ctx, r, func(impl *controller.Impl) controller.Options {
+		return controller.Options{
+			SkipStatusUpdates: true,
+		}
+	})
+
+	logging.FromContext(ctx).Info("Setting up event handlers")
+	kafkasourceInformer := kakfasourceinformer.Get(ctx)
+	kafkasourceInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
+
+	return impl
+}
diff --git a/pkg/source/mtadapter/kafkasource.go b/pkg/source/mtadapter/kafkasource.go
new file mode 100644
index 000000000..cd63f4616
--- /dev/null
+++ b/pkg/source/mtadapter/kafkasource.go
@@ -0,0 +1,56 @@
+/*
+Copyright 2020 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 mtadapter
+
+import (
+	"context"
+	"fmt"
+
+	"knative.dev/pkg/reconciler"
+
+	"knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	kafkasourcereconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
+)
+
+// Reconciler updates the internal Adapter cache kafkaSources
+type Reconciler struct {
+	mtadapter MTAdapter
+}
+
+// Check that our Reconciler implements ReconcileKind.
+var _ kafkasourcereconciler.Interface = (*Reconciler)(nil)
+
+// Check that our Reconciler implements FinalizeKind.
+var _ kafkasourcereconciler.Finalizer = (*Reconciler)(nil)
+
+func (r *Reconciler) ReconcileKind(ctx context.Context, source *v1beta1.KafkaSource) reconciler.Event {
+	if !source.Status.IsReady() {
+		return fmt.Errorf("warning: KafkaSource is not ready")
+	}
+
+	// Update the adapter state
+	r.mtadapter.Update(ctx, source)
+
+	return nil
+}
+
+func (r *Reconciler) FinalizeKind(ctx context.Context, source *v1beta1.KafkaSource) reconciler.Event {
+	// Update the adapter state
+	r.mtadapter.Remove(ctx, source)
+
+	return nil
+}
diff --git a/pkg/source/reconciler/mtsource/controller.go b/pkg/source/reconciler/mtsource/controller.go
new file mode 100644
index 000000000..7584ccf95
--- /dev/null
+++ b/pkg/source/reconciler/mtsource/controller.go
@@ -0,0 +1,65 @@
+/*
+Copyright 2019 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 mtsource
+
+import (
+	"context"
+
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/resolver"
+	"knative.dev/pkg/system"
+
+	"knative.dev/eventing/pkg/reconciler/source"
+
+	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"
+)
+
+func NewController(
+	ctx context.Context,
+	cmw configmap.Watcher,
+) *controller.Impl {
+	kafkaInformer := kafkainformer.Get(ctx)
+
+	c := &Reconciler{
+		KubeClientSet:  kubeclient.Get(ctx),
+		kafkaClientSet: kafkaclient.Get(ctx),
+		kafkaLister:    kafkaInformer.Lister(),
+		configs:        source.WatchConfigurations(ctx, component, cmw),
+	}
+
+	impl := kafkasource.NewImpl(ctx, c)
+
+	c.sinkResolver = resolver.NewURIResolver(ctx, impl.EnqueueKey)
+
+	// Use a different set of conditions
+	sourcesv1beta1.RegisterAlternateKafkaConditionSet(sourcesv1beta1.KafkaMTSourceCondSet)
+
+	c.scheduler = scheduler.NewStatefulSetScheduler(ctx, system.Namespace(), mtadapterName, c.vpodLister, impl.EnqueueKey)
+
+	logging.FromContext(ctx).Info("Setting up kafka event handlers")
+
+	kafkaInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
+
+	return impl
+}
diff --git a/pkg/source/reconciler/mtsource/doc.go b/pkg/source/reconciler/mtsource/doc.go
new file mode 100644
index 000000000..6f1bdb6cf
--- /dev/null
+++ b/pkg/source/reconciler/mtsource/doc.go
@@ -0,0 +1,18 @@
+/*
+Copyright 2019 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 source implements the Multi Tenant KafkaSource controller.
+package mtsource
diff --git a/pkg/source/reconciler/mtsource/kafkasource.go b/pkg/source/reconciler/mtsource/kafkasource.go
new file mode 100644
index 000000000..2f45715d7
--- /dev/null
+++ b/pkg/source/reconciler/mtsource/kafkasource.go
@@ -0,0 +1,151 @@
+/*
+Copyright 2019 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 mtsource
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/kubernetes"
+	"knative.dev/eventing/pkg/reconciler/source"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/logging"
+	pkgreconciler "knative.dev/pkg/reconciler"
+	"knative.dev/pkg/resolver"
+
+	"knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	reconcilerkafkasource "knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
+	listers "knative.dev/eventing-kafka/pkg/client/listers/sources/v1beta1"
+	"knative.dev/eventing-kafka/pkg/common/scheduler"
+)
+
+const (
+	component     = "kafkasource"
+	mtadapterName = "kafkasource-mt-adapter"
+)
+
+type Reconciler struct {
+	KubeClientSet  kubernetes.Interface
+	kafkaLister    listers.KafkaSourceLister
+	kafkaClientSet versioned.Interface
+
+	sinkResolver *resolver.URIResolver
+	configs      source.ConfigAccessor
+	scheduler    scheduler.Scheduler
+}
+
+// Check that our Reconciler implements Interface
+var _ reconcilerkafkasource.Interface = (*Reconciler)(nil)
+
+func (r *Reconciler) ReconcileKind(ctx context.Context, src *v1beta1.KafkaSource) pkgreconciler.Event {
+	src.Status.InitializeConditions()
+
+	if (src.Spec.Sink == duckv1.Destination{}) {
+		src.Status.MarkNoSink("SinkMissing", "")
+		return fmt.Errorf("spec.sink missing")
+	}
+
+	dest := src.Spec.Sink.DeepCopy()
+	if dest.Ref != nil {
+		// To call URIFromDestination(), dest.Ref must have a Namespace. If there is
+		// no Namespace defined in dest.Ref, we will use the Namespace of the source
+		// as the Namespace of dest.Ref.
+		if dest.Ref.Namespace == "" {
+			dest.Ref.Namespace = src.GetNamespace()
+		}
+	}
+	sinkURI, err := r.sinkResolver.URIFromDestinationV1(ctx, *dest, src)
+	if err != nil {
+		src.Status.MarkNoSink("NotFound", "")
+		return fmt.Errorf("getting sink URI: %v", err)
+	}
+	src.Status.MarkSink(sinkURI)
+
+	if val, ok := src.GetLabels()[v1beta1.KafkaKeyTypeLabel]; ok {
+		found := false
+		for _, allowed := range v1beta1.KafkaKeyTypeAllowed {
+			if allowed == val {
+				found = true
+			}
+		}
+		if !found {
+			src.Status.MarkKeyTypeIncorrect("IncorrectKafkaKeyTypeLabel", "Invalid value for %s: %s. Allowed: %v", v1beta1.KafkaKeyTypeLabel, val, v1beta1.KafkaKeyTypeAllowed)
+			logging.FromContext(ctx).Errorf("Invalid value for %s: %s. Allowed: %v", v1beta1.KafkaKeyTypeLabel, val, v1beta1.KafkaKeyTypeAllowed)
+			return errors.New("IncorrectKafkaKeyTypeLabel")
+		} else {
+			src.Status.MarkKeyTypeCorrect()
+		}
+	}
+
+	if err := r.reconcileMTReceiveAdapter(src); err != nil {
+		return err
+	}
+
+	src.Status.CloudEventAttributes = r.createCloudEventAttributes(src)
+
+	return nil
+}
+
+func (r *Reconciler) reconcileMTReceiveAdapter(src *v1beta1.KafkaSource) error {
+	placements, err := r.scheduler.Schedule(src)
+
+	// Update placements, even partial ones.
+	if placements != nil {
+		src.Status.Placement = placements
+	}
+
+	if err != nil {
+		src.Status.MarkNotScheduled("Unschedulable", err.Error())
+		return err // retrying...
+	}
+	src.Status.MarkScheduled()
+
+	// TODO: patch envvars
+	//return r.KubeClientSet.AppsV1().DaemonSets(system.Namespace()).Get(ctx, mtadapterName, metav1.GetOptions{})
+
+	return nil
+}
+
+func (r *Reconciler) vpodLister() ([]scheduler.VPod, error) {
+	sources, err := r.kafkaLister.List(labels.Everything())
+	if err != nil {
+		return nil, err
+	}
+	vpods := make([]scheduler.VPod, len(sources))
+	for i := 0; i < len(sources); i++ {
+		vpods[i] = sources[i]
+	}
+	return vpods, nil
+}
+
+func (r *Reconciler) createCloudEventAttributes(src *v1beta1.KafkaSource) []duckv1.CloudEventAttributes {
+	ceAttributes := make([]duckv1.CloudEventAttributes, 0, len(src.Spec.Topics))
+	for i := range src.Spec.Topics {
+		topics := strings.Split(src.Spec.Topics[i], ",")
+		for _, topic := range topics {
+			ceAttributes = append(ceAttributes, duckv1.CloudEventAttributes{
+				Type:   v1beta1.KafkaEventType,
+				Source: v1beta1.KafkaEventSource(src.Namespace, src.Name, topic),
+			})
+		}
+	}
+	return ceAttributes
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1/broker_types.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1/broker_types.go
index 8574e6280..f968fdc7e 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1/broker_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1/broker_types.go
@@ -74,8 +74,9 @@ type BrokerSpec struct {
 	// +optional
 	Config *duckv1.KReference `json:"config,omitempty"`
 
-	// Delivery is the delivery specification for Events within the Broker mesh.
-	// This includes things like retries, DLQ, etc.
+	// Delivery contains the delivery spec for each trigger
+	// to this Broker. Each trigger delivery spec, if any, overrides this
+	// global delivery spec.
 	// +optional
 	Delivery *eventingduckv1.DeliverySpec `json:"delivery,omitempty"`
 }
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1/trigger_types.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1/trigger_types.go
index 1db26c286..7e27a9cbe 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1/trigger_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1/trigger_types.go
@@ -22,6 +22,8 @@ import (
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
+
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 )
 
 const (
@@ -84,6 +86,10 @@ type TriggerSpec struct {
 	// Subscriber is the addressable that receives events from the Broker that pass the Filter. It
 	// is required.
 	Subscriber duckv1.Destination `json:"subscriber"`
+
+	// Delivery contains the delivery spec for this specific trigger.
+	// +optional
+	Delivery *eventingduckv1.DeliverySpec `json:"delivery,omitempty"`
 }
 
 type TriggerFilter struct {
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1/trigger_validation.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1/trigger_validation.go
index 895a1f7c2..4b367340a 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1/trigger_validation.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1/trigger_validation.go
@@ -69,6 +69,12 @@ func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(fe.ViaField("subscriber"))
 	}
 
+	if ts.Delivery != nil {
+		if de := ts.Delivery.Validate(ctx); de != nil {
+			errs = errs.Also(de.ViaField("delivery"))
+		}
+	}
+
 	return errs
 }
 
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1/zz_generated.deepcopy.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1/zz_generated.deepcopy.go
index ade414b8e..c1f4be248 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1/zz_generated.deepcopy.go
@@ -247,6 +247,11 @@ func (in *TriggerSpec) DeepCopyInto(out *TriggerSpec) {
 		(*in).DeepCopyInto(*out)
 	}
 	in.Subscriber.DeepCopyInto(&out.Subscriber)
+	if in.Delivery != nil {
+		in, out := &in.Delivery, &out.Delivery
+		*out = new(apisduckv1.DeliverySpec)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/trigger_conversion.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/trigger_conversion.go
index 5b03af797..f311d2ef8 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/trigger_conversion.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/trigger_conversion.go
@@ -20,12 +20,14 @@ import (
 	"context"
 	"fmt"
 
+	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	duckv1beta1 "knative.dev/eventing/pkg/apis/duck/v1beta1"
 	v1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/pkg/apis"
 )
 
 // ConvertTo implements apis.Convertible
-func (source *Trigger) ConvertTo(_ context.Context, to apis.Convertible) error {
+func (source *Trigger) ConvertTo(ctx context.Context, to apis.Convertible) error {
 	switch sink := to.(type) {
 	case *v1.Trigger:
 		sink.ObjectMeta = source.ObjectMeta
@@ -39,6 +41,12 @@ func (source *Trigger) ConvertTo(_ context.Context, to apis.Convertible) error {
 				sink.Spec.Filter.Attributes[k] = v
 			}
 		}
+		if source.Spec.Delivery != nil {
+			sink.Spec.Delivery = &duckv1.DeliverySpec{}
+			if err := source.Spec.Delivery.ConvertTo(ctx, sink.Spec.Delivery); err != nil {
+				return err
+			}
+		}
 		sink.Status.Status = source.Status.Status
 		sink.Status.SubscriberURI = source.Status.SubscriberURI
 		return nil
@@ -48,7 +56,7 @@ func (source *Trigger) ConvertTo(_ context.Context, to apis.Convertible) error {
 }
 
 // ConvertFrom implements apis.Convertible
-func (sink *Trigger) ConvertFrom(_ context.Context, from apis.Convertible) error {
+func (sink *Trigger) ConvertFrom(ctx context.Context, from apis.Convertible) error {
 	switch source := from.(type) {
 	case *v1.Trigger:
 		sink.ObjectMeta = source.ObjectMeta
@@ -63,6 +71,12 @@ func (sink *Trigger) ConvertFrom(_ context.Context, from apis.Convertible) error
 				Attributes: attributes,
 			}
 		}
+		if source.Spec.Delivery != nil {
+			sink.Spec.Delivery = &duckv1beta1.DeliverySpec{}
+			if err := sink.Spec.Delivery.ConvertFrom(ctx, source.Spec.Delivery); err != nil {
+				return err
+			}
+		}
 		sink.Status.Status = source.Status.Status
 		sink.Status.SubscriberURI = source.Status.SubscriberURI
 		return nil
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/trigger_types.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/trigger_types.go
index 914dab277..e4f0da375 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/trigger_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/trigger_types.go
@@ -22,6 +22,8 @@ import (
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
+
+	eventingduckv1beta1 "knative.dev/eventing/pkg/apis/duck/v1beta1"
 )
 
 const (
@@ -87,6 +89,10 @@ type TriggerSpec struct {
 	// Subscriber is the addressable that receives events from the Broker that pass the Filter. It
 	// is required.
 	Subscriber duckv1.Destination `json:"subscriber"`
+
+	// Delivery contains the delivery spec for this specific trigger.
+	// +optional
+	Delivery *eventingduckv1beta1.DeliverySpec `json:"delivery,omitempty"`
 }
 
 type TriggerFilter struct {
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/trigger_validation.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/trigger_validation.go
index 9c07ef724..b29e87014 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/trigger_validation.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/trigger_validation.go
@@ -65,6 +65,12 @@ func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(fe.ViaField("subscriber"))
 	}
 
+	if ts.Delivery != nil {
+		if de := ts.Delivery.Validate(ctx); de != nil {
+			errs = errs.Also(de.ViaField("delivery"))
+		}
+	}
+
 	return errs
 }
 
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/zz_generated.deepcopy.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/zz_generated.deepcopy.go
index c582712e4..fb5af442b 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/zz_generated.deepcopy.go
@@ -351,6 +351,11 @@ func (in *TriggerSpec) DeepCopyInto(out *TriggerSpec) {
 		(*in).DeepCopyInto(*out)
 	}
 	in.Subscriber.DeepCopyInto(&out.Subscriber)
+	if in.Delivery != nil {
+		in, out := &in.Delivery, &out.Delivery
+		*out = new(duckv1beta1.DeliverySpec)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake/fake.go
new file mode 100644
index 000000000..104f1e2c7
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake/fake.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	statefulset "knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset"
+	fake "knative.dev/pkg/client/injection/kube/informers/factory/fake"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = statefulset.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Apps().V1().StatefulSets()
+	return context.WithValue(ctx, statefulset.Key{}, inf), inf.Informer()
+}
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/statefulset.go b/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/statefulset.go
new file mode 100644
index 000000000..1f25874dd
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/statefulset.go
@@ -0,0 +1,52 @@
+/*
+Copyright 2020 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package statefulset
+
+import (
+	context "context"
+
+	v1 "k8s.io/client-go/informers/apps/v1"
+	factory "knative.dev/pkg/client/injection/kube/informers/factory"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformer(withInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := factory.Get(ctx)
+	inf := f.Apps().V1().StatefulSets()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1.StatefulSetInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch k8s.io/client-go/informers/apps/v1.StatefulSetInformer from context.")
+	}
+	return untyped.(v1.StatefulSetInformer)
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index eb6b703c6..52ee90907 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -477,6 +477,7 @@ golang.org/x/text/unicode/bidi
 golang.org/x/text/unicode/norm
 golang.org/x/text/width
 # golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e
+## explicit
 golang.org/x/time/rate
 # golang.org/x/tools v0.0.0-20201211185031-d93e913c1a58
 golang.org/x/tools/cmd/goimports
@@ -1086,9 +1087,10 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.20.1-0.20210130010931-b20c96b4df51
+# knative.dev/eventing v0.20.1-0.20210201132532-321a8399379c
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
+knative.dev/eventing/pkg/adapter/v2/test
 knative.dev/eventing/pkg/adapter/v2/util/crstatusevent
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/configs
@@ -1247,6 +1249,8 @@ knative.dev/pkg/client/injection/kube/informers/admissionregistration/v1/mutatin
 knative.dev/pkg/client/injection/kube/informers/admissionregistration/v1/validatingwebhookconfiguration
 knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment
 knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment/fake
+knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset
+knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/endpoints
 knative.dev/pkg/client/injection/kube/informers/core/v1/namespace
 knative.dev/pkg/client/injection/kube/informers/core/v1/secret


Test patch:
diff --git a/pkg/apis/sources/v1beta1/kafka_scheduling_test.go b/pkg/apis/sources/v1beta1/kafka_scheduling_test.go
new file mode 100644
index 000000000..d7e26d45a
--- /dev/null
+++ b/pkg/apis/sources/v1beta1/kafka_scheduling_test.go
@@ -0,0 +1,84 @@
+/*
+Copyright 2020 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 v1beta1
+
+import (
+	"reflect"
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/pointer"
+	"knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+)
+
+func TestScheduling(t *testing.T) {
+	testCases := map[string]struct {
+		source     KafkaSource
+		key        types.NamespacedName
+		vreplicas  int32
+		placements []v1alpha1.Placement
+	}{
+		"all empty": {
+			source:     KafkaSource{},
+			key:        types.NamespacedName{},
+			vreplicas:  int32(1),
+			placements: nil,
+		},
+		"all full": {
+			source: KafkaSource{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "asource",
+					Namespace: "anamespace",
+				},
+				Spec: KafkaSourceSpec{
+					Consumers: pointer.Int32Ptr(4),
+				},
+				Status: KafkaSourceStatus{
+					Placeable: v1alpha1.Placeable{Placement: []v1alpha1.Placement{
+						{PodName: "apod", VReplicas: 4},
+					}},
+				},
+			},
+			key: types.NamespacedName{
+				Namespace: "anamespace",
+				Name:      "asource",
+			},
+			vreplicas: int32(4),
+			placements: []v1alpha1.Placement{
+				{PodName: "apod", VReplicas: 4},
+			},
+		},
+	}
+
+	for n, tc := range testCases {
+		tc := tc
+		t.Run(n, func(t *testing.T) {
+			t.Parallel()
+
+			if !reflect.DeepEqual(tc.source.GetKey(), tc.key) {
+				t.Errorf("unexpected key (want %v, got %v)", tc.key, tc.source.GetKey())
+			}
+			if tc.source.GetVReplicas() != tc.vreplicas {
+				t.Errorf("unexpected vreplicas (want %d, got %d)", tc.vreplicas, tc.source.GetVReplicas())
+			}
+			if !reflect.DeepEqual(tc.source.GetPlacements(), tc.placements) {
+				t.Errorf("unexpected placements (want %v, got %v)", tc.placements, tc.source.GetPlacements())
+			}
+		})
+	}
+}
diff --git a/pkg/apis/sources/v1beta1/mttest/kafka_lifecycle_mt_test.go b/pkg/apis/sources/v1beta1/mttest/kafka_lifecycle_mt_test.go
new file mode 100644
index 000000000..d235e5748
--- /dev/null
+++ b/pkg/apis/sources/v1beta1/mttest/kafka_lifecycle_mt_test.go
@@ -0,0 +1,158 @@
+/*
+Copyright 2020 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 mttest
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	corev1 "k8s.io/api/core/v1"
+	"knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/pkg/apis"
+)
+
+func TestMTKafkaSourceStatusGetCondition(t *testing.T) {
+	v1beta1.RegisterAlternateKafkaConditionSet(v1beta1.KafkaMTSourceCondSet)
+	tests := []struct {
+		name      string
+		s         *v1beta1.KafkaSourceStatus
+		condQuery apis.ConditionType
+		want      *apis.Condition
+	}{{
+		name:      "uninitialized",
+		s:         &v1beta1.KafkaSourceStatus{},
+		condQuery: v1beta1.KafkaConditionReady,
+		want:      nil,
+	}, {
+		name: "initialized",
+		s: func() *v1beta1.KafkaSourceStatus {
+			s := &v1beta1.KafkaSourceStatus{}
+			s.InitializeConditions()
+			return s
+		}(),
+		condQuery: v1beta1.KafkaConditionReady,
+		want: &apis.Condition{
+			Type:   v1beta1.KafkaConditionReady,
+			Status: corev1.ConditionUnknown,
+		},
+	}, {
+		name: "mark scheduled",
+		s: func() *v1beta1.KafkaSourceStatus {
+			s := &v1beta1.KafkaSourceStatus{}
+			s.InitializeConditions()
+			s.MarkScheduled()
+			return s
+		}(),
+		condQuery: v1beta1.KafkaConditionReady,
+		want: &apis.Condition{
+			Type:   v1beta1.KafkaConditionReady,
+			Status: corev1.ConditionUnknown,
+		},
+	}, {
+		name: "mark sink",
+		s: func() *v1beta1.KafkaSourceStatus {
+			s := &v1beta1.KafkaSourceStatus{}
+			s.InitializeConditions()
+			s.MarkSink(apis.HTTP("example"))
+			return s
+		}(),
+		condQuery: v1beta1.KafkaConditionReady,
+		want: &apis.Condition{
+			Type:   v1beta1.KafkaConditionReady,
+			Status: corev1.ConditionUnknown,
+		},
+	}, {
+		name: "mark sink and scheduled",
+		s: func() *v1beta1.KafkaSourceStatus {
+			s := &v1beta1.KafkaSourceStatus{}
+			s.InitializeConditions()
+			s.MarkSink(apis.HTTP("uri://example"))
+			s.MarkScheduled()
+			return s
+		}(),
+		condQuery: v1beta1.KafkaConditionReady,
+		want: &apis.Condition{
+			Type:   v1beta1.KafkaConditionReady,
+			Status: corev1.ConditionTrue,
+		},
+	}, {
+		name: "mark sink and scheduled then no sink",
+		s: func() *v1beta1.KafkaSourceStatus {
+			s := &v1beta1.KafkaSourceStatus{}
+			s.InitializeConditions()
+			s.MarkSink(apis.HTTP("example"))
+			s.MarkScheduled()
+			s.MarkNoSink("Testing", "hi%s", "")
+			return s
+		}(),
+		condQuery: v1beta1.KafkaConditionReady,
+		want: &apis.Condition{
+			Type:    v1beta1.KafkaConditionReady,
+			Status:  corev1.ConditionFalse,
+			Reason:  "Testing",
+			Message: "hi",
+		},
+	}, {
+		name: "mark sink nil and scheduled",
+		s: func() *v1beta1.KafkaSourceStatus {
+			s := &v1beta1.KafkaSourceStatus{}
+			s.InitializeConditions()
+			s.MarkSink(nil)
+			s.MarkScheduled()
+			return s
+		}(),
+		condQuery: v1beta1.KafkaConditionReady,
+		want: &apis.Condition{
+			Type:    v1beta1.KafkaConditionReady,
+			Status:  corev1.ConditionUnknown,
+			Reason:  "SinkEmpty",
+			Message: "Sink has resolved to empty.",
+		},
+	}, {
+		name: "mark sink empty and scheduled then sink",
+		s: func() *v1beta1.KafkaSourceStatus {
+			s := &v1beta1.KafkaSourceStatus{}
+			s.InitializeConditions()
+			s.MarkSink(nil)
+			s.MarkScheduled()
+			s.MarkSink(apis.HTTP("example"))
+			return s
+		}(),
+		condQuery: v1beta1.KafkaConditionReady,
+		want: &apis.Condition{
+			Type:   v1beta1.KafkaConditionReady,
+			Status: corev1.ConditionTrue,
+		},
+	}}
+
+	for _, test := range tests {
+		test := test
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+			got := test.s.GetCondition(test.condQuery)
+			fmt.Println(test.s)
+			ignoreTime := cmpopts.IgnoreFields(apis.Condition{},
+				"LastTransitionTime", "Severity")
+			if diff := cmp.Diff(test.want, got, ignoreTime); diff != "" {
+
+				t.Errorf("unexpected condition (-want, +got) = %v", diff)
+			}
+		})
+	}
+}
diff --git a/pkg/common/consumer/consumer_handler_test.go b/pkg/common/consumer/consumer_handler_test.go
index f1e96919b..354869611 100644
--- a/pkg/common/consumer/consumer_handler_test.go
+++ b/pkg/common/consumer/consumer_handler_test.go
@@ -69,7 +69,7 @@ func (m *mockConsumerGroupSession) MarkMessage(msg *sarama.ConsumerMessage, meta
 }
 
 func (m *mockConsumerGroupSession) Context() context.Context {
-	return nil
+	return context.Background()
 }
 
 var _ sarama.ConsumerGroupSession = (*mockConsumerGroupSession)(nil)
diff --git a/pkg/common/scheduler/placement_test.go b/pkg/common/scheduler/placement_test.go
new file mode 100644
index 000000000..713c187b5
--- /dev/null
+++ b/pkg/common/scheduler/placement_test.go
@@ -0,0 +1,108 @@
+/*
+Copyright 2020 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 scheduler
+
+import (
+	"testing"
+
+	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+)
+
+func TestGetTotalVReplicas(t *testing.T) {
+	testCases := []struct {
+		name       string
+		placements []duckv1alpha1.Placement
+		vreplicas  int
+	}{
+		{
+			name:       "nil placements",
+			placements: nil,
+			vreplicas:  0,
+		},
+		{
+			name:       "empty placements",
+			placements: []duckv1alpha1.Placement{},
+			vreplicas:  0,
+		},
+		{
+			name:       "one placement",
+			placements: []duckv1alpha1.Placement{{PodName: "d", VReplicas: 2}},
+			vreplicas:  2,
+		},
+		{
+			name: "many placements",
+			placements: []duckv1alpha1.Placement{
+				{PodName: "d", VReplicas: 2},
+				{PodName: "d", VReplicas: 6},
+				{PodName: "d", VReplicas: 0}},
+			vreplicas: 8,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			vreplicas := GetTotalVReplicas(tc.placements)
+			if vreplicas != int32(tc.vreplicas) {
+				t.Errorf("got %d, want %d", vreplicas, tc.vreplicas)
+			}
+		})
+	}
+}
+
+func TestGetPlacementForPod(t *testing.T) {
+	ps1 := []duckv1alpha1.Placement{{PodName: "p", VReplicas: 2}}
+	ps2 := []duckv1alpha1.Placement{{PodName: "p", VReplicas: 2}, {PodName: "p2", VReplicas: 4}}
+	testCases := []struct {
+		name       string
+		podName    string
+		placements []duckv1alpha1.Placement
+		expected   *duckv1alpha1.Placement
+	}{
+		{
+			name:       "nil placements",
+			podName:    "p",
+			placements: nil,
+			expected:   nil,
+		},
+		{
+			name:       "empty placements",
+			podName:    "p",
+			placements: []duckv1alpha1.Placement{},
+			expected:   nil,
+		},
+		{
+			name:       "one placement",
+			placements: ps1,
+			podName:    "p",
+			expected:   &ps1[0],
+		}, {
+			name:       "mayne placements",
+			placements: ps2,
+			podName:    "p2",
+			expected:   &ps2[1],
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := GetPlacementForPod(tc.placements, tc.podName)
+			if got != tc.expected {
+				t.Errorf("got %v, want %v", got, tc.expected)
+			}
+		})
+	}
+}
diff --git a/pkg/common/scheduler/statefulset/scheduler_test.go b/pkg/common/scheduler/statefulset/scheduler_test.go
new file mode 100644
index 000000000..d1c062930
--- /dev/null
+++ b/pkg/common/scheduler/statefulset/scheduler_test.go
@@ -0,0 +1,187 @@
+/*
+Copyright 2020 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 statefulset
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"reflect"
+	"testing"
+	"time"
+
+	appsv1 "k8s.io/api/apps/v1"
+	autoscalingv1 "k8s.io/api/autoscaling/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	gtesting "k8s.io/client-go/testing"
+
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake"
+	"knative.dev/pkg/controller"
+	rectesting "knative.dev/pkg/reconciler/testing"
+
+	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+	"knative.dev/eventing-kafka/pkg/common/scheduler"
+	tscheduler "knative.dev/eventing-kafka/pkg/common/scheduler/testing"
+)
+
+const (
+	sfsName = "statefulset-name"
+)
+
+func TestStatefulsetScheduler(t *testing.T) {
+	ctx, cancel := setupFakeContext(t)
+
+	testCases := []struct {
+		name       string
+		vpods      []int32
+		replicas   int32
+		placements []duckv1alpha1.Placement
+		err        error
+	}{
+		{
+			name:       "no replicas",
+			vpods:      []int32{1},
+			replicas:   int32(0),
+			err:        scheduler.ErrNoReplicas,
+			placements: []duckv1alpha1.Placement{},
+		},
+		{
+			name:       "one vpod, one vreplicas",
+			vpods:      []int32{1},
+			replicas:   int32(1),
+			placements: []duckv1alpha1.Placement{{PodName: "statefulset-name-0", VReplicas: 1}},
+		},
+		{
+			name:       "one vpod, 3 vreplicas",
+			vpods:      []int32{3},
+			replicas:   int32(1),
+			placements: []duckv1alpha1.Placement{{PodName: "statefulset-name-0", VReplicas: 3}},
+		},
+		{
+			name:       "one vpod, 15 vreplicas, partial scheduling",
+			vpods:      []int32{15},
+			replicas:   int32(1),
+			err:        scheduler.ErrNotEnoughReplicas,
+			placements: []duckv1alpha1.Placement{{PodName: "statefulset-name-0", VReplicas: 10}},
+		},
+		{
+			name:     "one vpod, 15 vreplicas, full scheduling",
+			vpods:    []int32{15},
+			replicas: int32(2),
+			placements: []duckv1alpha1.Placement{
+				{PodName: "statefulset-name-0", VReplicas: 10},
+				{PodName: "statefulset-name-1", VReplicas: 5},
+			},
+		},
+	}
+
+	for i, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			vpodClient := tscheduler.NewVPodClient()
+			sfsNs := fmt.Sprint("ns-", i)
+
+			_, err := kubeclient.Get(ctx).AppsV1().StatefulSets(sfsNs).Create(ctx, makeStatefulset(ctx, sfsNs, sfsName, tc.replicas), metav1.CreateOptions{})
+			if err != nil {
+				t.Fatal("unexpected error", err)
+			}
+
+			s := NewStatefulSetScheduler(ctx, sfsNs, sfsName, vpodClient.List, nil).(*StatefulSetScheduler)
+
+			// Give some time for the informer to notify the scheduler
+			time.Sleep(500 * time.Millisecond)
+
+			func() {
+				s.lock.Lock()
+				defer s.lock.Unlock()
+				if s.replicas != tc.replicas {
+					t.Fatalf("expected number of statefulset replica to be %d (got %d)", tc.replicas, s.replicas)
+				}
+			}()
+
+			for i, vreplicas := range tc.vpods {
+				vpodName := fmt.Sprint("vpod-name-", i)
+				vpodNamespace := fmt.Sprint("vpod-ns-", i)
+
+				vpod := vpodClient.Create(vpodNamespace, vpodName, vreplicas, nil)
+				placements, err := s.Schedule(vpod)
+
+				if tc.err == nil && err != nil {
+					t.Fatal("unexpected error", err)
+				}
+
+				if tc.err != nil && err == nil {
+					t.Fatal("expected error, got none")
+				}
+
+				if !reflect.DeepEqual(placements, tc.placements) {
+					t.Errorf("got %v, want %v", placements, tc.placements)
+				}
+			}
+		})
+	}
+	cancel()
+	<-ctx.Done()
+}
+
+func makeStatefulset(ctx context.Context, ns, name string, replicas int32) *appsv1.StatefulSet {
+	obj := &appsv1.StatefulSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: ns,
+		},
+		Spec: appsv1.StatefulSetSpec{
+			Replicas: &replicas,
+		},
+		Status: appsv1.StatefulSetStatus{
+			Replicas: replicas,
+		},
+	}
+
+	kc := kubeclient.Get(ctx)
+	kc.PrependReactor("get", "statefulsets", func(action gtesting.Action) (handled bool, ret runtime.Object, err error) {
+		if action.GetSubresource() == "scale" {
+			return true, &autoscalingv1.Scale{
+				Spec: autoscalingv1.ScaleSpec{
+					Replicas: obj.Status.Replicas,
+				},
+			}, nil
+		}
+		return false, nil, nil
+	})
+
+	return obj
+}
+
+func setupFakeContext(t *testing.T) (context.Context, context.CancelFunc) {
+	ctx, cancel, informers := rectesting.SetupFakeContextWithCancel(t)
+	err := controller.StartInformers(ctx.Done(), informers...)
+	if err != nil {
+		t.Fatal("unexpected error", err)
+	}
+
+	kc := kubeclient.Get(ctx)
+	kc.PrependReactor("get", "statefulsets", func(action gtesting.Action) (handled bool, ret runtime.Object, err error) {
+		if action.GetSubresource() == "scale" {
+			return true, nil, errors.New("object not found")
+		}
+		return false, nil, nil
+	})
+
+	return ctx, cancel
+}
diff --git a/pkg/common/scheduler/statefulset/snapshot_test.go b/pkg/common/scheduler/statefulset/snapshot_test.go
new file mode 100644
index 000000000..f35ce2a82
--- /dev/null
+++ b/pkg/common/scheduler/statefulset/snapshot_test.go
@@ -0,0 +1,122 @@
+/*
+Copyright 2020 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 statefulset
+
+import (
+	"fmt"
+	"reflect"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"knative.dev/pkg/logging"
+
+	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+	tscheduler "knative.dev/eventing-kafka/pkg/common/scheduler/testing"
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+)
+
+func TestSnapshot(t *testing.T) {
+	ctx, cancel := setupFakeContext(t)
+
+	testCases := []struct {
+		name     string
+		vpods    [][]duckv1alpha1.Placement
+		replicas int32
+		expected Snapshot
+		err      error
+	}{
+		{
+			name:     "no vpods, no replicas",
+			vpods:    [][]duckv1alpha1.Placement{},
+			replicas: int32(0),
+			expected: Snapshot{free: map[string]int32{}},
+		},
+		{
+			name:     "no vpods, one replica",
+			vpods:    [][]duckv1alpha1.Placement{},
+			replicas: int32(1),
+			expected: Snapshot{free: map[string]int32{"statefulset-name-0": int32(10)}},
+		},
+		{
+			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)}},
+		},
+		{
+			name: "many vpods, many replicas",
+			vpods: [][]duckv1alpha1.Placement{
+				{{PodName: "statefulset-name-0", VReplicas: 1}, {PodName: "statefulset-name-2", VReplicas: 5}},
+				{{PodName: "statefulset-name-1", VReplicas: 2}},
+				{{PodName: "statefulset-name-1", VReplicas: 3}, {PodName: "statefulset-name-0", VReplicas: 1}},
+			},
+			replicas: int32(4),
+			expected: Snapshot{free: map[string]int32{
+				"statefulset-name-0": int32(8),
+				"statefulset-name-1": int32(5),
+				"statefulset-name-2": int32(5),
+				"statefulset-name-3": int32(10)}},
+		},
+	}
+
+	for i, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			vpodClient := tscheduler.NewVPodClient()
+			sfsNs := fmt.Sprint("ns-", i)
+
+			_, err := kubeclient.Get(ctx).AppsV1().StatefulSets(sfsNs).Create(ctx, makeStatefulset(ctx, sfsNs, sfsName, tc.replicas), metav1.CreateOptions{})
+			if err != nil {
+				t.Fatal("unexpected error", err)
+			}
+
+			s := NewStatefulSetScheduler(ctx, sfsNs, sfsName, vpodClient.List, nil).(*StatefulSetScheduler)
+
+			// Give some time for the informer to notify the scheduler
+			time.Sleep(500 * time.Millisecond)
+
+			func() {
+				s.lock.Lock()
+				defer s.lock.Unlock()
+				if s.replicas != tc.replicas {
+					t.Fatalf("expected number of statefulset replica to be %d (got %d)", tc.replicas, s.replicas)
+				}
+			}()
+
+			for i, placements := range tc.vpods {
+				vpodName := fmt.Sprint("vpod-name-", i)
+				vpodNamespace := fmt.Sprint("vpod-ns-", i)
+
+				vpodClient.Create(vpodNamespace, vpodName, 1, placements)
+
+			}
+
+			shot, err := s.snapshot(logging.FromContext(ctx))
+			if err != nil {
+				t.Fatal("unexpected error", err)
+			}
+
+			if !reflect.DeepEqual(*shot, tc.expected) {
+				t.Errorf("got %v, want %v", shot, tc.expected)
+			}
+		})
+	}
+
+	cancel()
+	<-ctx.Done()
+}
diff --git a/pkg/common/scheduler/testing/client.go b/pkg/common/scheduler/testing/client.go
new file mode 100644
index 000000000..2d51251c8
--- /dev/null
+++ b/pkg/common/scheduler/testing/client.go
@@ -0,0 +1,47 @@
+/*
+Copyright 2020 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 testing
+
+import (
+	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+	"knative.dev/eventing-kafka/pkg/common/scheduler"
+)
+
+type VPodClient struct {
+	store  *VPodStore
+	lister scheduler.VPodLister
+}
+
+func NewVPodClient() *VPodClient {
+	store := newVPodStore()
+	return &VPodClient{
+		store:  store,
+		lister: newVPodLister(store),
+	}
+}
+
+func (s *VPodClient) Create(ns, name string, vreplicas int32, placements []duckv1alpha1.Placement) scheduler.VPod {
+	vpod := newVPod(ns, name, vreplicas, placements)
+	s.store.lock.Lock()
+	defer s.store.lock.Unlock()
+	s.store.vpods = append(s.store.vpods, vpod)
+	return vpod
+}
+
+func (s *VPodClient) List() ([]scheduler.VPod, error) {
+	return s.lister()
+}
diff --git a/pkg/common/scheduler/testing/store.go b/pkg/common/scheduler/testing/store.go
new file mode 100644
index 000000000..ecd28ad2c
--- /dev/null
+++ b/pkg/common/scheduler/testing/store.go
@@ -0,0 +1,42 @@
+/*
+Copyright 2020 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 testing
+
+import (
+	"sync"
+
+	"knative.dev/eventing-kafka/pkg/common/scheduler"
+)
+
+type VPodStore struct {
+	vpods []scheduler.VPod
+	lock  sync.Mutex
+}
+
+func newVPodStore() *VPodStore {
+	return &VPodStore{}
+}
+
+func newVPodLister(store *VPodStore) scheduler.VPodLister {
+	return func() ([]scheduler.VPod, error) {
+		store.lock.Lock()
+		defer store.lock.Unlock()
+		dst := make([]scheduler.VPod, len(store.vpods))
+		copy(dst, store.vpods)
+		return dst, nil
+	}
+}
diff --git a/pkg/common/scheduler/testing/vpod.go b/pkg/common/scheduler/testing/vpod.go
new file mode 100644
index 000000000..4751fa46a
--- /dev/null
+++ b/pkg/common/scheduler/testing/vpod.go
@@ -0,0 +1,51 @@
+/*
+Copyright 2020 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 testing
+
+import (
+	"k8s.io/apimachinery/pkg/types"
+	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+)
+
+type sampleVPod struct {
+	key        types.NamespacedName
+	vreplicas  int32
+	placements []duckv1alpha1.Placement
+}
+
+func newVPod(ns, name string, vreplicas int32, placements []duckv1alpha1.Placement) *sampleVPod {
+	return &sampleVPod{
+		key: types.NamespacedName{
+			Namespace: ns,
+			Name:      name,
+		},
+		vreplicas:  vreplicas,
+		placements: placements,
+	}
+}
+
+func (d *sampleVPod) GetKey() types.NamespacedName {
+	return d.key
+}
+
+func (d *sampleVPod) GetVReplicas() int32 {
+	return d.vreplicas
+}
+
+func (d *sampleVPod) GetPlacements() []duckv1alpha1.Placement {
+	return d.placements
+}
diff --git a/pkg/source/adapter/adapter_benchmark_test.go b/pkg/source/adapter/adapter_benchmark_test.go
index a0f2271db..9c58b0237 100644
--- a/pkg/source/adapter/adapter_benchmark_test.go
+++ b/pkg/source/adapter/adapter_benchmark_test.go
@@ -73,7 +73,7 @@ func BenchmarkHandle(b *testing.B) {
 	statsReporter, _ := source.NewStatsReporter()
 
 	a := &Adapter{
-		config: &adapterConfig{
+		config: &AdapterConfig{
 			EnvConfig: adapter.EnvConfig{
 				Sink:      sinkUrl,
 				Namespace: "test",
diff --git a/pkg/source/adapter/adapter_test.go b/pkg/source/adapter/adapter_test.go
index 9c18caeb0..5b81670ca 100644
--- a/pkg/source/adapter/adapter_test.go
+++ b/pkg/source/adapter/adapter_test.go
@@ -341,7 +341,7 @@ func TestPostMessage_ServeHTTP_binary_mode(t *testing.T) {
 			}
 
 			a := &Adapter{
-				config: &adapterConfig{
+				config: &AdapterConfig{
 					EnvConfig: adapter.EnvConfig{
 						Sink:      sinkServer.URL,
 						Namespace: "test",
diff --git a/pkg/source/mtadapter/adapter_test.go b/pkg/source/mtadapter/adapter_test.go
new file mode 100644
index 000000000..94690790e
--- /dev/null
+++ b/pkg/source/mtadapter/adapter_test.go
@@ -0,0 +1,147 @@
+/*
+Copyright 2020 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 mtadapter
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	pkgtesting "knative.dev/pkg/reconciler/testing"
+	"knative.dev/pkg/source"
+
+	"knative.dev/eventing/pkg/adapter/v2"
+	adaptertest "knative.dev/eventing/pkg/adapter/v2/test"
+	"knative.dev/eventing/pkg/kncloudevents"
+
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+)
+
+var (
+	runningAdapterChan  = make(chan *sampleAdapter)
+	stoppingAdapterChan = make(chan *sampleAdapter)
+)
+
+const (
+	podName = "sample-podname"
+)
+
+func TestUpdateRemoveSources(t *testing.T) {
+	ctx, _ := pkgtesting.SetupFakeContext(t)
+	ctx, cancelAdapter := context.WithCancel(ctx)
+
+	env := &AdapterConfig{PodName: podName}
+	ceClient := adaptertest.NewTestClient()
+
+	adapter := newAdapter(ctx, env, ceClient, newSampleAdapter).(*Adapter)
+
+	adapterStopped := make(chan bool)
+	go func() {
+		err := adapter.Start(ctx)
+		if err != nil {
+			t.Error("Unexpected error ", err)
+		}
+		adapterStopped <- true
+	}()
+
+	adapter.Update(ctx, &sourcesv1beta1.KafkaSource{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-name",
+			Namespace: "test-ns",
+		},
+		Spec: sourcesv1beta1.KafkaSourceSpec{},
+		Status: sourcesv1beta1.KafkaSourceStatus{
+			Placeable: duckv1alpha1.Placeable{
+				Placement: []duckv1alpha1.Placement{
+					{PodName: podName, VReplicas: int32(1)},
+				}},
+		},
+	})
+
+	if _, ok := adapter.sources["test-ns/test-name"]; !ok {
+		t.Error(`Expected adapter to contain "test-ns/test-name"`)
+	}
+
+	select {
+	case a := <-runningAdapterChan:
+		if !a.running {
+			t.Error("Expected adapter to be running")
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Error("sub-adapter failed to start after 100 ms")
+	}
+
+	adapter.Remove(ctx, &sourcesv1beta1.KafkaSource{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-name",
+			Namespace: "test-ns",
+		},
+		Spec:   sourcesv1beta1.KafkaSourceSpec{},
+		Status: sourcesv1beta1.KafkaSourceStatus{},
+	})
+
+	if _, ok := adapter.sources["test-ns/test-name"]; ok {
+		t.Error(`Expected adapter to not contain "test-ns/test-name"`)
+	}
+
+	select {
+	case a := <-stoppingAdapterChan:
+		if a.running {
+			t.Error("Expected adapter to not be running")
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Error("sub-adapter failed to stop after 100 ms")
+	}
+
+	// Make sure the adapter is still running
+	select {
+	case <-ctx.Done():
+		t.Error("Expected adapter to be running")
+	default:
+	}
+
+	cancelAdapter()
+
+	select {
+	case <-adapterStopped:
+	case <-time.After(2 * time.Second):
+		t.Error("adapter failed to stop after 2 seconds")
+	}
+}
+
+type sampleAdapter struct {
+	running bool
+}
+
+func newSampleAdapter(ctx context.Context, env adapter.EnvConfigAccessor, adapter *kncloudevents.HTTPMessageSender, reporter source.StatsReporter) adapter.MessageAdapter {
+	return &sampleAdapter{}
+}
+
+func (d *sampleAdapter) Start(ctx context.Context) error {
+	d.running = true
+	runningAdapterChan <- d
+
+	<-ctx.Done()
+
+	d.running = false
+	stoppingAdapterChan <- d
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/adapter/v2/test/test_client.go b/vendor/knative.dev/eventing/pkg/adapter/v2/test/test_client.go
new file mode 100644
index 000000000..395c66c33
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/adapter/v2/test/test_client.go
@@ -0,0 +1,92 @@
+/*
+Copyright 2020 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 test
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/cloudevents/sdk-go/v2/event"
+	"github.com/cloudevents/sdk-go/v2/protocol"
+	"github.com/cloudevents/sdk-go/v2/protocol/http"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+)
+
+type TestCloudEventsClient struct {
+	lock  sync.Mutex
+	sent  []cloudevents.Event
+	delay time.Duration
+}
+
+var _ cloudevents.Client = (*TestCloudEventsClient)(nil)
+
+func (c *TestCloudEventsClient) Send(ctx context.Context, out event.Event) protocol.Result {
+	if c.delay > 0 {
+		time.Sleep(c.delay)
+	}
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	// TODO: improve later.
+	c.sent = append(c.sent, out)
+	return http.NewResult(200, "%w", protocol.ResultACK)
+}
+
+func (c *TestCloudEventsClient) Request(ctx context.Context, out event.Event) (*event.Event, protocol.Result) {
+	if c.delay > 0 {
+		time.Sleep(c.delay)
+	}
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	// TODO: improve later.
+	c.sent = append(c.sent, out)
+	return nil, http.NewResult(200, "%w", protocol.ResultACK)
+}
+
+func (c *TestCloudEventsClient) StartReceiver(ctx context.Context, fn interface{}) error {
+	// TODO: improve later.
+	<-ctx.Done()
+	return nil
+}
+
+func (c *TestCloudEventsClient) Reset() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.sent = make([]cloudevents.Event, 0)
+}
+
+func (c *TestCloudEventsClient) Sent() []cloudevents.Event {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	r := make([]cloudevents.Event, len(c.sent))
+	for i := range c.sent {
+		r[i] = c.sent[i]
+	}
+	return r
+}
+
+func NewTestClient() *TestCloudEventsClient {
+	return NewTestClientWithDelay(0)
+}
+
+func NewTestClientWithDelay(delay time.Duration) *TestCloudEventsClient {
+	c := &TestCloudEventsClient{
+		sent:  make([]cloudevents.Event, 0),
+		delay: delay,
+	}
+	return c
+}