Repository: knative-sandbox/eventing-kafka
Pull request #349 (2021-02-01 15:49:23)

Add steven0711dong to the owner list 
Add myself to owner list

google-cla[bot]: 
Thanks for your pull request. It looks like this may be your first contribution to a Google open source project (if not, look below for help). Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

:memo: **Please visit <https://cla.developers.google.com/> to sign.**

Once you've signed (or fixed any issues), please reply here with `@googlebot I signed it!` and we'll verify it.

----

#### What to do if you already signed the CLA

##### Individual signers

*   It's possible we don't have your GitHub username or you're using a different email address on your commit. Check [your existing CLA data](https://cla.developers.google.com/clas) and verify that your [email is set on your git commits](https://help.github.com/articles/setting-your-email-in-git/).

##### Corporate signers

*   Your company has a Point of Contact who decides which employees are authorized to participate. Ask your POC to be added to the group of authorized contributors. If you don't know who your Point of Contact is, direct the Google project maintainer to [go/cla#troubleshoot](http://go/cla#troubleshoot) ([Public version](https://opensource.google/docs/cla/#troubleshoot)).
*   The email used to register you as an authorized contributor must be the email used for the Git commit. Check [your existing CLA data](https://cla.developers.google.com/clas) and verify that your [email is set on your git commits](https://help.github.com/articles/setting-your-email-in-git/).
*   The email used to register you as an authorized contributor must also be [attached to your GitHub account](https://github.com/settings/emails).
		

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fknative-sandbox%2Feventing-kafka%2Fpull%2F349) for more info**.

<!-- need_sender_cla -->
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/349#" title="Author self-approved">steven0711dong</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **phamilton** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @phamilton` 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":["phamilton"]} -->
knative-prow-robot: Welcome @steven0711dong! It looks like this is your first PR to knative-sandbox/eventing-kafka 🎉
knative-prow-robot: Hi @steven0711dong. Thanks for your PR.

I'm waiting for a [knative-sandbox](https://github.com/orgs/knative-sandbox/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/knative-sandbox/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

<details>

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

knative-prow-robot: The following users are mentioned in OWNERS file(s) but are untrusted for the following reasons. One way to make the user trusted is to add them as [members]() of the knative-sandbox org. You can then trigger verification by writing `/verify-owners` in a comment.
- steven0711dong
  - User is not a member of the org. User is not a collaborator. Satisfy at least one of these conditions to make the user trusted.
  - OWNERS
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/349?src=pr&el=h1) Report
> Merging [#349](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/349?src=pr&el=desc) (c911f0f) 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/349/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/349?src=pr&el=tree)

```diff
@@           Coverage Diff           @@
##           master     #349   +/-   ##
=======================================
  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/349?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/349?src=pr&el=footer). Last update [ca62759...ca6f8cf](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/349?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

knative-prow-robot: The following users are mentioned in OWNERS file(s) but are untrusted for the following reasons. One way to make the user trusted is to add them as [members]() of the knative-sandbox org. You can then trigger verification by writing `/verify-owners` in a comment.
- steven0711dong
  - User is not a member of the org. User is not a collaborator. Satisfy at least one of these conditions to make the user trusted.
  - OWNERS
steven0711dong: /verify-owners
knative-prow-robot: The following users are mentioned in OWNERS file(s) but are untrusted for the following reasons. One way to make the user trusted is to add them as [members]() of the knative-sandbox org. You can then trigger verification by writing `/verify-owners` in a comment.
- steven0711dong
  - User is not a member of the org. User is not a collaborator. Satisfy at least one of these conditions to make the user trusted.
  - OWNERS
knative-prow-robot: The following users are mentioned in OWNERS file(s) but are untrusted for the following reasons. One way to make the user trusted is to add them as [members]() of the knative-sandbox org. You can then trigger verification by writing `/verify-owners` in a comment.
- steven0711dong
  - User is not a member of the org. User is not a collaborator. Satisfy at least one of these conditions to make the user trusted.

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..633cda055 100644
--- a/go.mod
+++ b/go.mod
@@ -28,6 +28,7 @@ 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
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/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..a23709218 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
@@ -1089,6 +1090,7 @@ k8s.io/utils/trace
 # knative.dev/eventing v0.20.1-0.20210130010931-b20c96b4df51
 ## 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
+}