Repository: knative-sandbox/eventing-kafka-broker
Pull request #1794 (2022-01-24 20:02:59)

[data-plane]: Bump fabric8.kubernetes.version from 5.11.1 to 5.12.0 in /data-plane
Bumps `fabric8.kubernetes.version` from 5.11.1 to 5.12.0.
Updates `kubernetes-client` from 5.11.1 to 5.12.0
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/fabric8io/kubernetes-client/releases">kubernetes-client's releases</a>.</em></p>
<blockquote>
<h2>5.12.0 (2022-01-24)</h2>
<h4>Bugs</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3683">#3683</a>: Handle JsonNode fields by adding x-kubernetes-preserve-unknown-fields</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3697">#3697</a>: addresses response that aren't closed by interceptors that issue new requests</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3255">#3255</a>: adding basic crud mock resourceVersion support - the field will be set and updated, but not utilized by list/watch queries</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3568">#3568</a>: Pod file upload fails if the path is <code>/</code></li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3588">#3588</a>: <code>openshift-server-mock</code> is not listed in dependencyManagement in main pom</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3648">#3648</a>: <code>Serialization.unmarshal</code> fails to deserialize YAML with single document in presence of document delimiter(<code>---</code>)</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3679">#3679</a>: output additionalProperties field with correct value type for map-like fields (CRD Generator)</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3671">#3671</a>: HTTP(s) Proxy port is not defaulted or validated</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3712">#3712</a>: properly return the full resource name for resources with empty group</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3761">#3761</a>: Extension Jar packages don't contain the META-INF/jandex.idx index file</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3763">#3763</a>: A Java Long should generate a field of type integer in the CRD</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3769">#3769</a>: Fix for ClassCastException from SchemaFrom</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3756">#3756</a> prevent modifications by standard operations to user objects</li>
</ul>
<h4>Improvements</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3674">#3674</a>: allows the connect and websocket timeouts to apply to watches instead of a hardcoded timeout</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3651">#3651</a>: Introduce SchemaFrom annotation as escape hatch (CRD Generator)</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3587">#3587</a>: adding inform support for limit/batch fetching</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3734">#3734</a>: extract static finalizer validation method</li>
</ul>
<h4>Dependency Upgrade</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3637">#3637</a>: Update Fabric8 Kubernetes Model to v1.23.0</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3670">#3670</a>: Point CamelK Extension model to latest released version v1.7.0</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3725">#3725</a>: Bump sundrio to v0.50.3</li>
</ul>
<h4>New Features</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3721">#3721</a>: Add support for uploading file via InputStream</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3234">#3234</a>: Allow specifying specific localhost while port-forwarding</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3506">#3506</a>: Add support for Open Cluster Management extension</li>
</ul>
<h2>5.11.2 (2022-01-05)</h2>
<h4>Bugs</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3653">#3653</a>: SnakeYAML uses only standard Java types</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3697">#3697</a>: addresses response that aren't closed by interceptors that issue new requests</li>
</ul>
</blockquote>
</details>
<details>
<summary>Changelog</summary>
<p><em>Sourced from <a href="https://github.com/fabric8io/kubernetes-client/blob/master/CHANGELOG.md">kubernetes-client's changelog</a>.</em></p>
<blockquote>
<h3>5.12.0 (2022-01-24)</h3>
<h4>Bugs</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3683">#3683</a>: Handle JsonNode fields by adding x-kubernetes-preserve-unknown-fields</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3697">#3697</a>: addresses response that aren't closed by interceptors that issue new requests</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3255">#3255</a>: adding basic crud mock resourceVersion support - the field will be set and updated, but not utilized by list/watch queries</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3568">#3568</a>: Pod file upload fails if the path is <code>/</code></li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3588">#3588</a>: <code>openshift-server-mock</code> is not listed in dependencyManagement in main pom</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3648">#3648</a>: <code>Serialization.unmarshal</code> fails to deserialize YAML with single document in presence of document delimiter(<code>---</code>)</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3679">#3679</a>: output additionalProperties field with correct value type for map-like fields (CRD Generator)</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3671">#3671</a>: HTTP(s) Proxy port is not defaulted or validated</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3712">#3712</a>: properly return the full resource name for resources with empty group</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3761">#3761</a>: Extension Jar packages don't contain the META-INF/jandex.idx index file</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3763">#3763</a>: A Java Long should generate a field of type integer in the CRD</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3769">#3769</a>: Fix for ClassCastException from SchemaFrom</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3756">#3756</a> prevent modifications by standard operations to user objects</li>
</ul>
<h4>Improvements</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3674">#3674</a>: allows the connect and websocket timeouts to apply to watches instead of a hardcoded timeout</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3651">#3651</a>: Introduce SchemaFrom annotation as escape hatch (CRD Generator)</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3587">#3587</a>: adding inform support for limit/batch fetching</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3734">#3734</a>: extract static finalizer validation method</li>
</ul>
<h4>Dependency Upgrade</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3637">#3637</a>: Update Fabric8 Kubernetes Model to v1.23.0</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3670">#3670</a>: Point CamelK Extension model to latest released version v1.7.0</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3725">#3725</a>: Bump sundrio to v0.50.3</li>
</ul>
<h4>New Features</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3721">#3721</a>: Add support for uploading file via InputStream</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3234">#3234</a>: Allow specifying specific localhost while port-forwarding</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3506">#3506</a>: Add support for Open Cluster Management extension</li>
</ul>
<h3>5.10.2 (2022-01-07)</h3>
<h4>Bugs</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3653">#3653</a>: SnakeYAML uses only standard Java types</li>
</ul>
<h3>5.11.2 (2022-01-05)</h3>
<h4>Bugs</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3653">#3653</a>: SnakeYAML uses only standard Java types</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3697">#3697</a>: addresses response that aren't closed by interceptors that issue new requests</li>
</ul>
<h3>5.8.1 (2022-01-05)</h3>
<h4>Bugs</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3653">#3653</a>: SnakeYAML uses only standard Java types</li>
</ul>
<h3>5.7.4 (2022-01-05)</h3>
<!-- raw HTML omitted -->
</blockquote>
<p>... (truncated)</p>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/8ab32f978d4755c4b813ce5604a293c5d8bf5c3c"><code>8ab32f9</code></a> [RELEASE] Updated project version to v5.12.0</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/26c08847d1251b2e840ce5833bbe05e43ee77fa6"><code>26c0884</code></a> fix(kubernetes-client) : HTTP(s) Proxy Port is not defaulted or validated (<a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3">#3</a>...</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/987c9e3ab38ee310411885f0f17086290249bbe3"><code>987c9e3</code></a> arquliian dep</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/7a46c26726c531c854e1b3aa60ba8ea7b120047c"><code>7a46c26</code></a> Add OpenShift Hive and Installer Model Modules and DSL support</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/9f1a5690029d21bd5a5cc2ce73566c0f1da5d742"><code>9f1a569</code></a> Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3506">#3506</a>: Add Extension for Open Cluster Management</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/4bc001e860de949191fb23b3b84d53671f227071"><code>4bc001e</code></a> Extension Model Generator should convert byte[] to String</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/658b2f88df298a098a521a3318f1e61ade316b0d"><code>658b2f8</code></a> Add GroupKind from Kubernetes Apimachinery metav1 to Kubernetes Model Core</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/c4ac742c8fff03ca126dc5b2ecb226f0437f1156"><code>c4ac742</code></a> review: reusable osgi.include.resources.default property</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/257632fc54146a4e6f826baeadac0c5161e991e0"><code>257632f</code></a> fix(extensions): Extension Jars don't contain <code>META-INF/jandex.idx</code> file (<a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3761">#3761</a>)</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/5b7442c9ebdc8d0226014cd760ee9e6663f550c5"><code>5b7442c</code></a> chore: add missing changelog entries</li>
<li>Additional commits viewable in <a href="https://github.com/fabric8io/kubernetes-client/compare/v5.11.1...v5.12.0">compare view</a></li>
</ul>
</details>
<br />

Updates `kubernetes-server-mock` from 5.11.1 to 5.12.0
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/fabric8io/kubernetes-client/releases">kubernetes-server-mock's releases</a>.</em></p>
<blockquote>
<h2>5.12.0 (2022-01-24)</h2>
<h4>Bugs</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3683">#3683</a>: Handle JsonNode fields by adding x-kubernetes-preserve-unknown-fields</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3697">#3697</a>: addresses response that aren't closed by interceptors that issue new requests</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3255">#3255</a>: adding basic crud mock resourceVersion support - the field will be set and updated, but not utilized by list/watch queries</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3568">#3568</a>: Pod file upload fails if the path is <code>/</code></li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3588">#3588</a>: <code>openshift-server-mock</code> is not listed in dependencyManagement in main pom</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3648">#3648</a>: <code>Serialization.unmarshal</code> fails to deserialize YAML with single document in presence of document delimiter(<code>---</code>)</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3679">#3679</a>: output additionalProperties field with correct value type for map-like fields (CRD Generator)</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3671">#3671</a>: HTTP(s) Proxy port is not defaulted or validated</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3712">#3712</a>: properly return the full resource name for resources with empty group</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3761">#3761</a>: Extension Jar packages don't contain the META-INF/jandex.idx index file</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3763">#3763</a>: A Java Long should generate a field of type integer in the CRD</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3769">#3769</a>: Fix for ClassCastException from SchemaFrom</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3756">#3756</a> prevent modifications by standard operations to user objects</li>
</ul>
<h4>Improvements</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3674">#3674</a>: allows the connect and websocket timeouts to apply to watches instead of a hardcoded timeout</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3651">#3651</a>: Introduce SchemaFrom annotation as escape hatch (CRD Generator)</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3587">#3587</a>: adding inform support for limit/batch fetching</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3734">#3734</a>: extract static finalizer validation method</li>
</ul>
<h4>Dependency Upgrade</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3637">#3637</a>: Update Fabric8 Kubernetes Model to v1.23.0</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3670">#3670</a>: Point CamelK Extension model to latest released version v1.7.0</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3725">#3725</a>: Bump sundrio to v0.50.3</li>
</ul>
<h4>New Features</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3721">#3721</a>: Add support for uploading file via InputStream</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3234">#3234</a>: Allow specifying specific localhost while port-forwarding</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3506">#3506</a>: Add support for Open Cluster Management extension</li>
</ul>
<h2>5.11.2 (2022-01-05)</h2>
<h4>Bugs</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3653">#3653</a>: SnakeYAML uses only standard Java types</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3697">#3697</a>: addresses response that aren't closed by interceptors that issue new requests</li>
</ul>
</blockquote>
</details>
<details>
<summary>Changelog</summary>
<p><em>Sourced from <a href="https://github.com/fabric8io/kubernetes-client/blob/master/CHANGELOG.md">kubernetes-server-mock's changelog</a>.</em></p>
<blockquote>
<h3>5.12.0 (2022-01-24)</h3>
<h4>Bugs</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3683">#3683</a>: Handle JsonNode fields by adding x-kubernetes-preserve-unknown-fields</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3697">#3697</a>: addresses response that aren't closed by interceptors that issue new requests</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3255">#3255</a>: adding basic crud mock resourceVersion support - the field will be set and updated, but not utilized by list/watch queries</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3568">#3568</a>: Pod file upload fails if the path is <code>/</code></li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3588">#3588</a>: <code>openshift-server-mock</code> is not listed in dependencyManagement in main pom</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3648">#3648</a>: <code>Serialization.unmarshal</code> fails to deserialize YAML with single document in presence of document delimiter(<code>---</code>)</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3679">#3679</a>: output additionalProperties field with correct value type for map-like fields (CRD Generator)</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3671">#3671</a>: HTTP(s) Proxy port is not defaulted or validated</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3712">#3712</a>: properly return the full resource name for resources with empty group</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3761">#3761</a>: Extension Jar packages don't contain the META-INF/jandex.idx index file</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3763">#3763</a>: A Java Long should generate a field of type integer in the CRD</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3769">#3769</a>: Fix for ClassCastException from SchemaFrom</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3756">#3756</a> prevent modifications by standard operations to user objects</li>
</ul>
<h4>Improvements</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3674">#3674</a>: allows the connect and websocket timeouts to apply to watches instead of a hardcoded timeout</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3651">#3651</a>: Introduce SchemaFrom annotation as escape hatch (CRD Generator)</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3587">#3587</a>: adding inform support for limit/batch fetching</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3734">#3734</a>: extract static finalizer validation method</li>
</ul>
<h4>Dependency Upgrade</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3637">#3637</a>: Update Fabric8 Kubernetes Model to v1.23.0</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3670">#3670</a>: Point CamelK Extension model to latest released version v1.7.0</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3725">#3725</a>: Bump sundrio to v0.50.3</li>
</ul>
<h4>New Features</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3721">#3721</a>: Add support for uploading file via InputStream</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3234">#3234</a>: Allow specifying specific localhost while port-forwarding</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3506">#3506</a>: Add support for Open Cluster Management extension</li>
</ul>
<h3>5.10.2 (2022-01-07)</h3>
<h4>Bugs</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3653">#3653</a>: SnakeYAML uses only standard Java types</li>
</ul>
<h3>5.11.2 (2022-01-05)</h3>
<h4>Bugs</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3653">#3653</a>: SnakeYAML uses only standard Java types</li>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3697">#3697</a>: addresses response that aren't closed by interceptors that issue new requests</li>
</ul>
<h3>5.8.1 (2022-01-05)</h3>
<h4>Bugs</h4>
<ul>
<li>Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3653">#3653</a>: SnakeYAML uses only standard Java types</li>
</ul>
<h3>5.7.4 (2022-01-05)</h3>
<!-- raw HTML omitted -->
</blockquote>
<p>... (truncated)</p>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/8ab32f978d4755c4b813ce5604a293c5d8bf5c3c"><code>8ab32f9</code></a> [RELEASE] Updated project version to v5.12.0</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/26c08847d1251b2e840ce5833bbe05e43ee77fa6"><code>26c0884</code></a> fix(kubernetes-client) : HTTP(s) Proxy Port is not defaulted or validated (<a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3">#3</a>...</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/987c9e3ab38ee310411885f0f17086290249bbe3"><code>987c9e3</code></a> arquliian dep</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/7a46c26726c531c854e1b3aa60ba8ea7b120047c"><code>7a46c26</code></a> Add OpenShift Hive and Installer Model Modules and DSL support</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/9f1a5690029d21bd5a5cc2ce73566c0f1da5d742"><code>9f1a569</code></a> Fix <a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3506">#3506</a>: Add Extension for Open Cluster Management</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/4bc001e860de949191fb23b3b84d53671f227071"><code>4bc001e</code></a> Extension Model Generator should convert byte[] to String</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/658b2f88df298a098a521a3318f1e61ade316b0d"><code>658b2f8</code></a> Add GroupKind from Kubernetes Apimachinery metav1 to Kubernetes Model Core</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/c4ac742c8fff03ca126dc5b2ecb226f0437f1156"><code>c4ac742</code></a> review: reusable osgi.include.resources.default property</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/257632fc54146a4e6f826baeadac0c5161e991e0"><code>257632f</code></a> fix(extensions): Extension Jars don't contain <code>META-INF/jandex.idx</code> file (<a href="https://github-redirect.dependabot.com/fabric8io/kubernetes-client/issues/3761">#3761</a>)</li>
<li><a href="https://github.com/fabric8io/kubernetes-client/commit/5b7442c9ebdc8d0226014cd760ee9e6663f550c5"><code>5b7442c</code></a> chore: add missing changelog entries</li>
<li>Additional commits viewable in <a href="https://github.com/fabric8io/kubernetes-client/compare/v5.11.1...v5.12.0">compare view</a></li>
</ul>
</details>
<br />


Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>

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-broker/pull/1794#" title="Author self-approved">dependabot[bot]</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **aliok** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @aliok` 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-broker).

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/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":["aliok"]} -->
knative-prow-robot: Hi @dependabot[bot]. 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-broker).

<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: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1794#" title="Author self-approved">dependabot[bot]</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **pierdipi** after the PR has been reviewed.

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

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/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":["pierdipi"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1794?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1794](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1794?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (24a964f) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/0058cf3afcfe1332cfc892c9e2cf8a79db00d4f8?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (0058cf3) will **increase** coverage by `0.01%`.
> The diff coverage is `n/a`.

> :exclamation: Current head 24a964f differs from pull request most recent head 0617ef8. Consider uploading reports for the commit 0617ef8 to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1794/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1794?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##               main    #1794      +/-   ##
============================================
+ Coverage     68.63%   68.65%   +0.01%     
  Complexity      617      617              
============================================
  Files           129      129              
  Lines          5334     5334              
  Branches        178      178              
============================================
+ Hits           3661     3662       +1     
+ Misses         1382     1381       -1     
  Partials        291      291              
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.16% <ø> (-0.10%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1794?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...tcher/impl/consumer/UnorderedConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1794/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9Vbm9yZGVyZWRDb25zdW1lclZlcnRpY2xlLmphdmE=) | `91.30% <0.00%> (-8.70%)` | :arrow_down: |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1794/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `76.82% <0.00%> (-2.44%)` | :arrow_down: |
| [control-plane/pkg/prober/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1794/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2NhY2hlLmdv) | `100.00% <0.00%> (+4.76%)` | :arrow_up: |
| [control-plane/pkg/prober/prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1794/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2Jlci5nbw==) | `75.00% <0.00%> (+15.00%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1794?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1794?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [0058cf3...0617ef8](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1794?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1794#" title="Author self-approved">dependabot[bot]</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1794#pullrequestreview-862075013" title="Approved">matzew</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)~~ [matzew]

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":[]} -->
knative-prow-robot: @dependabot[bot]: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 0617ef89b1426c8eaa6d099efaf85b0998188b1f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1794/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1485920825025499136) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1794). [Your PR dashboard](https://gubernator.knative.dev/pr/dependabot[bot]).

<details>

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1794#" title="Author self-approved">dependabot[bot]</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1794#pullrequestreview-862075013" title="Approved">matzew</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1794#pullrequestreview-862246604" title="Approved">pierDipi</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)~~ [matzew,pierDipi]

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":[]} -->
knative-prow-robot: @dependabot[bot]: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 3ee4480a2b506b0dfec3f629072f466d2cd11167 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1794/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1485961014263418880) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1794). [Your PR dashboard](https://gubernator.knative.dev/pr/dependabot[bot]).

<details>

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

Patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
index 4693af4db..e6078edab 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
@@ -17,6 +17,8 @@
 package v1alpha1
 
 import (
+	"strings"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -168,6 +170,20 @@ func (cg *ConsumerGroup) IsReady() bool {
 		cg.GetConditionSet().Manage(cg.GetStatus()).IsHappy()
 }
 
+// GetUserFacingResourceRef gets the resource reference to the user-facing resources
+// that are backed by this ConsumerGroup using the OwnerReference list.
+func (cg *ConsumerGroup) GetUserFacingResourceRef() *metav1.OwnerReference {
+	for i, or := range cg.OwnerReferences {
+		// TODO hardcoded resource kinds.
+		if strings.EqualFold(or.Kind, "trigger") ||
+			strings.EqualFold(or.Kind, "kafkasource") ||
+			strings.EqualFold(or.Kind, "kafkachannel") {
+			return &cg.OwnerReferences[i]
+		}
+	}
+	return nil
+}
+
 func (cg *ConsumerGroup) IsNotScheduled() bool {
 	// We want to return true when:
 	// - the condition isn't present, or
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
index 6bac11ace..ec5c7d83c 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
@@ -17,6 +17,8 @@
 package v1alpha1
 
 import (
+	"strings"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	bindings "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
@@ -213,3 +215,14 @@ func (c *Consumer) IsReady() bool {
 
 // ConsumerOption is a functional option for Consumer.
 type ConsumerOption func(consumer *Consumer)
+
+// GetConsumerGroup gets the resource reference to the ConsumerGroup
+// using the OwnerReference list.
+func (c *Consumer) GetConsumerGroup() *metav1.OwnerReference {
+	for i, or := range c.OwnerReferences {
+		if strings.EqualFold(or.Kind, ConsumerGroupGroupVersionKind.Kind) {
+			return &c.OwnerReferences[i]
+		}
+	}
+	return nil
+}
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index 40fef589c..062e08a2f 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -19,12 +19,10 @@ package consumer
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"knative.dev/pkg/apis"
@@ -220,19 +218,7 @@ func reconcileCEOverrides(c *kafkainternals.Consumer) *contract.CloudEventOverri
 
 func (r Reconciler) reconcileUserFacingResourceRef(c *kafkainternals.Consumer) (*contract.Reference, error) {
 
-	// Get ConsumerGroup from the owner references list.
-	var cOr *metav1.OwnerReference
-	for i, or := range c.OwnerReferences {
-		if strings.ToLower(or.Kind) == kafkainternals.ConsumerGroupGroupVersionKind.Kind {
-			cOr = &c.OwnerReferences[i]
-			break
-		}
-	}
-	if cOr == nil {
-		return nil, fmt.Errorf("failed to find %s in owner references", kafkainternals.ConsumerGroupGroupVersionKind.Kind)
-	}
-
-	cg, err := r.ConsumerGroupLister.ConsumerGroups(c.GetNamespace()).Get(cOr.Name)
+	cg, err := r.ConsumerGroupLister.ConsumerGroups(c.GetNamespace()).Get(c.GetConsumerGroup().Name)
 	if err != nil && !apierrors.IsNotFound(err) {
 		return nil, fmt.Errorf("failed to get %s: %w", kafkainternals.ConsumerGroupGroupVersionKind.Kind, err)
 	}
@@ -240,18 +226,7 @@ func (r Reconciler) reconcileUserFacingResourceRef(c *kafkainternals.Consumer) (
 		return nil, nil
 	}
 
-	// Get user facing resource from the owner references list.
-	var cgOr *metav1.OwnerReference
-	for i, or := range cg.OwnerReferences {
-		k := strings.ToLower(or.Kind)
-		// TODO hardcoded resource kinds.
-		if k == "trigger" || k == "kafkasource" || k == "kafkachannel" {
-			cgOr = &cg.OwnerReferences[i]
-			break
-		}
-	}
-
-	userFacingResource := cgOr
+	userFacingResource := cg.GetUserFacingResourceRef()
 	ref := &contract.Reference{
 		Uuid:      string(userFacingResource.UID),
 		Namespace: c.GetNamespace(),
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index 7d27f9aba..6445c606f 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -39,8 +39,10 @@ import (
 	kafkainternalslisters "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
 )
 
+type schedulerFunc func(s string) scheduler.Scheduler
+
 type Reconciler struct {
-	Scheduler       scheduler.Scheduler
+	SchedulerFunc   schedulerFunc
 	ConsumerLister  kafkainternalslisters.ConsumerLister
 	InternalsClient internalv1alpha1.InternalV1alpha1Interface
 
@@ -167,7 +169,7 @@ func (r Reconciler) finalizeConsumer(ctx context.Context, consumer *kafkainterna
 }
 
 func (r Reconciler) schedule(cg *kafkainternals.ConsumerGroup) error {
-	placements, err := r.Scheduler.Schedule(cg)
+	placements, err := r.SchedulerFunc(cg.GetUserFacingResourceRef().Kind).Schedule(cg)
 	if err != nil {
 		return cg.MarkScheduleConsumerFailed("Schedule", err)
 	}
diff --git a/control-plane/pkg/reconciler/consumergroup/controller.go b/control-plane/pkg/reconciler/consumergroup/controller.go
new file mode 100644
index 000000000..9b33e9f71
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumergroup/controller.go
@@ -0,0 +1,216 @@
+/*
+ * Copyright 2022 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 consumergroup
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"go.uber.org/multierr"
+	"go.uber.org/zap"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apiserver/pkg/storage/names"
+	kafkainformer "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource"
+	sources "knative.dev/eventing-kafka/pkg/client/listers/sources/v1beta1"
+	"knative.dev/eventing/pkg/scheduler"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	nodeinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/node"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/system"
+
+	statefulsetscheduler "knative.dev/eventing/pkg/scheduler/statefulset"
+
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	internalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumer"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup"
+	cgreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumergroup"
+)
+
+const (
+	// KafkaSourceScheduler is the key for the scheduler map for any KafkaSource.
+	// Keep this constant lowercase.
+	KafkaSourceScheduler = "kafkasource"
+	// KafkaTriggerScheduler is the key for the scheduler map for any Kafka Trigger.
+	// Keep this constant lowercase.
+	KafkaTriggerScheduler = "trigger"
+	// KafkaChannelScheduler is the key for the scheduler map for any KafkaChannel.
+	// Keep this constant lowercase.
+	KafkaChannelScheduler = "kafkachannel"
+
+	// ConfigKafkaSchedulerName is the name of the ConfigMap to configure the scheduler.
+	ConfigKafkaSchedulerName = "config-kafka-scheduler"
+	// ConfigKafkaDeSchedulerName is the name of the ConfigMap to configure the descheduler.
+	ConfigKafkaDeSchedulerName = "config-kafka-descheduler"
+)
+
+type SchedulerConfig struct {
+	StatefulSetName     string
+	RefreshPeriod       time.Duration
+	Capacity            int32
+	SchedulerPolicyType scheduler.SchedulerPolicyType
+	SchedulerPolicy     *scheduler.SchedulerPolicy
+	DeSchedulerPolicy   *scheduler.SchedulerPolicy
+}
+
+func NewController(ctx context.Context) *controller.Impl {
+
+	// TODO(pierDipi) use env variables to configure these.
+	c := SchedulerConfig{
+		RefreshPeriod:       2 * time.Minute,
+		Capacity:            50,
+		SchedulerPolicyType: "",
+		SchedulerPolicy:     schedulerPolicyFromConfigMapOrFail(ctx, ConfigKafkaSchedulerName),
+		DeSchedulerPolicy:   schedulerPolicyFromConfigMapOrFail(ctx, ConfigKafkaDeSchedulerName),
+	}
+
+	kafkaLister := kafkainformer.Get(ctx).Lister()
+	schedulers := map[string]scheduler.Scheduler{
+		KafkaSourceScheduler: createKafkaSourceScheduler(ctx, c, kafkaLister),
+		// TODO(pierDipi) Add Kafka Trigger and KafkaChannel schedulers.
+	}
+
+	r := &Reconciler{
+		SchedulerFunc:   func(s string) scheduler.Scheduler { return schedulers[strings.ToLower(s)] },
+		ConsumerLister:  consumer.Get(ctx).Lister(),
+		InternalsClient: internalsclient.Get(ctx).InternalV1alpha1(),
+		NameGenerator:   names.SimpleNameGenerator,
+		SystemNamespace: system.Namespace(),
+	}
+
+	impl := cgreconciler.NewImpl(ctx, r)
+
+	consumergroup.Get(ctx).Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
+	consumer.Get(ctx).Informer().AddEventHandler(controller.HandleAll(enqueueConsumerGroupFromConsumer(impl.EnqueueKey)))
+
+	return impl
+}
+
+func enqueueConsumerGroupFromConsumer(enqueue func(name types.NamespacedName)) func(obj interface{}) {
+	return func(obj interface{}) {
+		c := obj.(*kafkainternals.Consumer)
+		or := c.GetConsumerGroup()
+		if or != nil {
+			enqueue(types.NamespacedName{Namespace: c.GetNamespace(), Name: or.Name})
+		}
+	}
+}
+
+func createKafkaSourceScheduler(ctx context.Context, c SchedulerConfig, kafkaLister sources.KafkaSourceLister) scheduler.Scheduler {
+	return createStatefulSetScheduler(
+		ctx,
+		SchedulerConfig{
+			StatefulSetName:     "kafka-source-dispatcher",
+			RefreshPeriod:       c.RefreshPeriod,
+			Capacity:            c.Capacity,
+			SchedulerPolicyType: c.SchedulerPolicyType,
+			SchedulerPolicy:     c.SchedulerPolicy,
+			DeSchedulerPolicy:   c.DeSchedulerPolicy,
+		},
+		func() ([]scheduler.VPod, error) {
+			kafkaSources, err := kafkaLister.List(labels.Everything())
+			if err != nil {
+				return nil, err
+			}
+			vpods := make([]scheduler.VPod, len(kafkaSources))
+			for i := 0; i < len(kafkaSources); i++ {
+				vpods[i] = kafkaSources[i]
+			}
+			return vpods, nil
+		},
+	)
+}
+
+func createStatefulSetScheduler(ctx context.Context, c SchedulerConfig, lister scheduler.VPodLister) scheduler.Scheduler {
+	return statefulsetscheduler.NewScheduler(
+		ctx,
+		system.Namespace(),
+		c.StatefulSetName,
+		lister,
+		c.RefreshPeriod,
+		c.Capacity,
+		c.SchedulerPolicyType,
+		// TODO add nodes to controller cluster role
+		nodeinformer.Get(ctx).Lister(),
+		nil, // TODO No evictor, is this ok?
+		c.SchedulerPolicy,
+		c.DeSchedulerPolicy,
+	)
+}
+
+// schedulerPolicyFromConfigMapOrFail reads predicates and priorities data from configMap
+func schedulerPolicyFromConfigMapOrFail(ctx context.Context, configMapName string) *scheduler.SchedulerPolicy {
+	p, err := schedulerPolicyFromConfigMap(ctx, configMapName)
+	if err != nil {
+		logging.FromContext(ctx).Fatal(zap.Error(err))
+	}
+	return p
+}
+
+// schedulerPolicyFromConfigMap reads predicates and priorities data from configMap
+func schedulerPolicyFromConfigMap(ctx context.Context, configMapName string) (*scheduler.SchedulerPolicy, error) {
+	policyConfigMap, err := kubeclient.Get(ctx).CoreV1().ConfigMaps(system.Namespace()).Get(ctx, configMapName, metav1.GetOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("couldn't get scheduler policy config map %s/%s: %v", system.Namespace(), configMapName, err)
+	}
+
+	logger := logging.FromContext(ctx).
+		Desugar().
+		With(zap.String("configmap", configMapName))
+	policy := &scheduler.SchedulerPolicy{}
+
+	preds, found := policyConfigMap.Data["predicates"]
+	if !found {
+		return nil, fmt.Errorf("missing policy config map %s/%s value at key predicates", system.Namespace(), configMapName)
+	}
+	if err := json.NewDecoder(strings.NewReader(preds)).Decode(&policy.Predicates); err != nil {
+		return nil, fmt.Errorf("invalid policy %v: %v", preds, err)
+	}
+
+	priors, found := policyConfigMap.Data["priorities"]
+	if !found {
+		return nil, fmt.Errorf("missing policy config map value at key priorities")
+	}
+	if err := json.NewDecoder(strings.NewReader(priors)).Decode(&policy.Priorities); err != nil {
+		return nil, fmt.Errorf("invalid policy %v: %v", preds, err)
+	}
+
+	if errs := validatePolicy(policy); errs != nil {
+		return nil, multierr.Combine(err)
+	}
+
+	logger.Info("Schedulers policy registration", zap.Any("policy", policy))
+
+	return policy, nil
+}
+
+func validatePolicy(policy *scheduler.SchedulerPolicy) []error {
+	var validationErrors []error
+
+	for _, priority := range policy.Priorities {
+		if priority.Weight < scheduler.MinWeight || priority.Weight > scheduler.MaxWeight {
+			validationErrors = append(validationErrors, fmt.Errorf("priority %s should have a positive weight applied to it or it has overflown", priority.Name))
+		}
+	}
+	return validationErrors
+}
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 7efab1a74..1aaabe7c3 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -43,28 +43,28 @@ Lists of 158 third-party dependencies.
      (The Apache Software License, Version 2.0) cloudevents-sql (io.cloudevents:cloudevents-sql:2.3.0 - https://cloudevents.github.io/sdk-java/cloudevents-sql/)
      (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.8.0.Final - https://debezium.io/debezium-parent/debezium-api)
      (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.8.0.Final - https://debezium.io/debezium-parent/debezium-core)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Java Client (io.fabric8:kubernetes-client:5.11.1 - http://fabric8.io/kubernetes-client/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Admission Registration, Authentication and Authorization (io.fabric8:kubernetes-model-admissionregistration:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-admissionregistration/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: API Extensions (io.fabric8:kubernetes-model-apiextensions:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apiextensions/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Apps (io.fabric8:kubernetes-model-apps:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apps/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Autoscaling (io.fabric8:kubernetes-model-autoscaling:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-autoscaling/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Batch (io.fabric8:kubernetes-model-batch:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-batch/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Certificates (io.fabric8:kubernetes-model-certificates:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-certificates/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Common (io.fabric8:kubernetes-model-common:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-common/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Coordination (io.fabric8:kubernetes-model-coordination:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-coordination/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Core (io.fabric8:kubernetes-model-core:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-core/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Discovery (io.fabric8:kubernetes-model-discovery:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-discovery/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Events (io.fabric8:kubernetes-model-events:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-events/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Extensions (io.fabric8:kubernetes-model-extensions:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-extensions/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: FlowControl (io.fabric8:kubernetes-model-flowcontrol:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-flowcontrol/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Metrics (io.fabric8:kubernetes-model-metrics:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-metrics/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Networking (io.fabric8:kubernetes-model-networking:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-networking/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Node (io.fabric8:kubernetes-model-node:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-node/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Policy (io.fabric8:kubernetes-model-policy:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-policy/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: RBAC (io.fabric8:kubernetes-model-rbac:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-rbac/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Scheduling (io.fabric8:kubernetes-model-scheduling:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-scheduling/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Storage Class (io.fabric8:kubernetes-model-storageclass:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-storageclass/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Server Mock (io.fabric8:kubernetes-server-mock:5.11.1 - http://fabric8.io/kubernetes-server-mock/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Java Client (io.fabric8:kubernetes-client:5.12.0 - http://fabric8.io/kubernetes-client/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Admission Registration, Authentication and Authorization (io.fabric8:kubernetes-model-admissionregistration:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-admissionregistration/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: API Extensions (io.fabric8:kubernetes-model-apiextensions:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apiextensions/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Apps (io.fabric8:kubernetes-model-apps:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apps/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Autoscaling (io.fabric8:kubernetes-model-autoscaling:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-autoscaling/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Batch (io.fabric8:kubernetes-model-batch:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-batch/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Certificates (io.fabric8:kubernetes-model-certificates:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-certificates/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Common (io.fabric8:kubernetes-model-common:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-common/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Coordination (io.fabric8:kubernetes-model-coordination:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-coordination/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Core (io.fabric8:kubernetes-model-core:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-core/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Discovery (io.fabric8:kubernetes-model-discovery:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-discovery/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Events (io.fabric8:kubernetes-model-events:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-events/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Extensions (io.fabric8:kubernetes-model-extensions:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-extensions/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: FlowControl (io.fabric8:kubernetes-model-flowcontrol:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-flowcontrol/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Metrics (io.fabric8:kubernetes-model-metrics:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-metrics/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Networking (io.fabric8:kubernetes-model-networking:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-networking/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Node (io.fabric8:kubernetes-model-node:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-node/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Policy (io.fabric8:kubernetes-model-policy:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-policy/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: RBAC (io.fabric8:kubernetes-model-rbac:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-rbac/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Scheduling (io.fabric8:kubernetes-model-scheduling:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-scheduling/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Storage Class (io.fabric8:kubernetes-model-storageclass:5.12.0 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-storageclass/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Server Mock (io.fabric8:kubernetes-server-mock:5.12.0 - http://fabric8.io/kubernetes-server-mock/)
      (Apache License, Version 2.0) Fabric8 :: Mock Web Server (io.fabric8:mockwebserver:0.2.2 - http://fabric8.io/)
      (The Apache Software License, Version 2.0) zjsonpatch (io.fabric8:zjsonpatch:0.3.0 - https://github.com/fabric8io/zjsonpatch/)
      (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.8.2 - https://github.com/micrometer-metrics/micrometer)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 0dbfd92ea..eaddaf405 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -54,7 +54,7 @@
     <awaitility.version>4.1.1</awaitility.version>
     <junit.jupiter.version>5.8.2</junit.jupiter.version>
     <mokito.junit.jupiter.version>4.3.0</mokito.junit.jupiter.version>
-    <fabric8.kubernetes.version>5.11.1</fabric8.kubernetes.version>
+    <fabric8.kubernetes.version>5.12.0</fabric8.kubernetes.version>
     <kafka.version>2.8.1</kafka.version>
     <debezium.version>1.8.0.Final</debezium.version>
   </properties>
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/factory/registry.go b/vendor/knative.dev/eventing/pkg/scheduler/factory/registry.go
new file mode 100644
index 000000000..518a91056
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/factory/registry.go
@@ -0,0 +1,88 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package factory
+
+import (
+	"fmt"
+
+	state "knative.dev/eventing/pkg/scheduler/state"
+)
+
+// RegistryFP is a collection of all available filter plugins.
+type RegistryFP map[string]state.FilterPlugin
+
+// RegistrySP is a collection of all available scoring plugins.
+type RegistrySP map[string]state.ScorePlugin
+
+var (
+	FilterRegistry = make(RegistryFP)
+	ScoreRegistry  = make(RegistrySP)
+)
+
+// Register adds a new plugin to the registry. If a plugin with the same name
+// exists, it returns an error.
+func RegisterFP(name string, factory state.FilterPlugin) error {
+	if _, ok := FilterRegistry[name]; ok {
+		return fmt.Errorf("a filter plugin named %v already exists", name)
+	}
+	FilterRegistry[name] = factory
+	return nil
+}
+
+// Unregister removes an existing plugin from the registry. If no plugin with
+// the provided name exists, it returns an error.
+func UnregisterFP(name string) error {
+	if _, ok := FilterRegistry[name]; !ok {
+		return fmt.Errorf("no filter plugin named %v exists", name)
+	}
+	delete(FilterRegistry, name)
+	return nil
+}
+
+func GetFilterPlugin(name string) (state.FilterPlugin, error) {
+	if f, exist := FilterRegistry[name]; exist {
+		return f, nil
+	}
+	return nil, fmt.Errorf("no fitler plugin named %v exists", name)
+}
+
+// Register adds a new plugin to the registry. If a plugin with the same name
+// exists, it returns an error.
+func RegisterSP(name string, factory state.ScorePlugin) error {
+	if _, ok := ScoreRegistry[name]; ok {
+		return fmt.Errorf("a score plugin named %v already exists", name)
+	}
+	ScoreRegistry[name] = factory
+	return nil
+}
+
+// Unregister removes an existing plugin from the registry. If no plugin with
+// the provided name exists, it returns an error.
+func UnregisterSP(name string) error {
+	if _, ok := ScoreRegistry[name]; !ok {
+		return fmt.Errorf("no score plugin named %v exists", name)
+	}
+	delete(ScoreRegistry, name)
+	return nil
+}
+
+func GetScorePlugin(name string) (state.ScorePlugin, error) {
+	if f, exist := ScoreRegistry[name]; exist {
+		return f, nil
+	}
+	return nil, fmt.Errorf("no score plugin named %v exists", name)
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/availabilitynodepriority/availability_node_priority.go b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/availabilitynodepriority/availability_node_priority.go
new file mode 100644
index 000000000..7c83d43df
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/availabilitynodepriority/availability_node_priority.go
@@ -0,0 +1,117 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package availabilitynodepriority
+
+import (
+	"context"
+	"encoding/json"
+	"math"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/pkg/scheduler/factory"
+	state "knative.dev/eventing/pkg/scheduler/state"
+	"knative.dev/pkg/logging"
+)
+
+// AvailabilityNodePriority is a score plugin that favors pods that create an even spread of resources across nodes for HA
+type AvailabilityNodePriority struct {
+}
+
+// Verify AvailabilityNodePriority Implements ScorePlugin Interface
+var _ state.ScorePlugin = &AvailabilityNodePriority{}
+
+// Name of the plugin
+const Name = state.AvailabilityNodePriority
+
+const (
+	ErrReasonInvalidArg    = "invalid arguments"
+	ErrReasonNoResource    = "node does not exist"
+	ErrReasonNotEnoughPods = "pods not enough to satisfy node availability"
+)
+
+func init() {
+	factory.RegisterSP(Name, &AvailabilityNodePriority{})
+}
+
+// Name returns name of the plugin
+func (pl *AvailabilityNodePriority) Name() string {
+	return Name
+}
+
+// Score invoked at the score extension point. The "score" returned in this function is higher for nodes that create an even spread across nodes.
+func (pl *AvailabilityNodePriority) Score(ctx context.Context, args interface{}, states *state.State, feasiblePods []int32, key types.NamespacedName, podID int32) (uint64, *state.Status) {
+	logger := logging.FromContext(ctx).With("Score", pl.Name())
+	var score uint64 = 0
+
+	spreadArgs, ok := args.(string)
+	if !ok {
+		logger.Errorf("Scoring args %v for priority %q are not valid", args, pl.Name())
+		return 0, state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	skewVal := state.AvailabilityNodePriorityArgs{}
+	decoder := json.NewDecoder(strings.NewReader(spreadArgs))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&skewVal); err != nil {
+		return 0, state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	if states.Replicas > 0 { //need at least a pod to compute spread
+		var skew int32
+
+		//Need to check if there is at least one pod in every node to satisfy HA
+		if !state.SatisfyNodeAvailability(states.SchedulablePods, states) {
+			return 0, state.NewStatus(state.Unschedulable, ErrReasonNotEnoughPods)
+		}
+
+		_, nodeName, err := states.GetPodInfo(state.PodNameFromOrdinal(states.StatefulSetName, podID))
+		if err != nil {
+			return score, state.NewStatus(state.Error, ErrReasonNoResource)
+		}
+
+		currentReps := states.NodeSpread[key][nodeName]   //get #vreps on this node
+		for otherNodeName := range states.NodeToZoneMap { //compare with #vreps on other nodes
+			if otherNodeName != nodeName {
+				otherReps := states.NodeSpread[key][otherNodeName]
+				if skew = (currentReps + 1) - otherReps; skew < 0 {
+					skew = skew * int32(-1)
+				}
+
+				//logger.Infof("Current Node %v with %d and Other Node %v with %d causing skew %d", nodeName, currentReps, otherNodeName, otherReps, skew)
+				if skew > skewVal.MaxSkew {
+					logger.Infof("Pod %d in node %v will cause an uneven node spread %v with other node %v", podID, nodeName, states.NodeSpread[key], otherNodeName)
+				}
+				score = score + uint64(skew)
+			}
+		}
+
+		score = math.MaxUint64 - score //lesser skews get higher score
+	}
+
+	return score, state.NewStatus(state.Success)
+}
+
+// ScoreExtensions of the Score plugin.
+func (pl *AvailabilityNodePriority) ScoreExtensions() state.ScoreExtensions {
+	return pl
+}
+
+// NormalizeScore invoked after scoring all pods.
+func (pl *AvailabilityNodePriority) NormalizeScore(ctx context.Context, states *state.State, scores state.PodScoreList) *state.Status {
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/availabilityzonepriority/availability_zone_priority.go b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/availabilityzonepriority/availability_zone_priority.go
new file mode 100644
index 000000000..bcefa1d37
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/availabilityzonepriority/availability_zone_priority.go
@@ -0,0 +1,121 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package availabilityzonepriority
+
+import (
+	"context"
+	"encoding/json"
+	"math"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/pkg/scheduler/factory"
+	state "knative.dev/eventing/pkg/scheduler/state"
+	"knative.dev/pkg/logging"
+)
+
+// AvailabilityZonePriority is a score plugin that favors pods that create an even spread of resources across zones for HA
+type AvailabilityZonePriority struct {
+}
+
+// Verify AvailabilityZonePriority Implements ScorePlugin Interface
+var _ state.ScorePlugin = &AvailabilityZonePriority{}
+
+// Name of the plugin
+const Name = state.AvailabilityZonePriority
+
+const (
+	ErrReasonInvalidArg    = "invalid arguments"
+	ErrReasonNoResource    = "zone does not exist"
+	ErrReasonNotEnoughPods = "pods not enough to satisfy zone availability"
+)
+
+func init() {
+	factory.RegisterSP(Name, &AvailabilityZonePriority{})
+}
+
+// Name returns name of the plugin
+func (pl *AvailabilityZonePriority) Name() string {
+	return Name
+}
+
+// Score invoked at the score extension point. The "score" returned in this function is higher for zones that create an even spread across zones.
+func (pl *AvailabilityZonePriority) Score(ctx context.Context, args interface{}, states *state.State, feasiblePods []int32, key types.NamespacedName, podID int32) (uint64, *state.Status) {
+	logger := logging.FromContext(ctx).With("Score", pl.Name())
+	var score uint64 = 0
+
+	spreadArgs, ok := args.(string)
+	if !ok {
+		logger.Errorf("Scoring args %v for priority %q are not valid", args, pl.Name())
+		return 0, state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	skewVal := state.AvailabilityZonePriorityArgs{}
+	decoder := json.NewDecoder(strings.NewReader(spreadArgs))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&skewVal); err != nil {
+		return 0, state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	if states.Replicas > 0 { //need at least a pod to compute spread
+		var skew int32
+		zoneMap := make(map[string]struct{})
+		for _, zoneName := range states.NodeToZoneMap {
+			zoneMap[zoneName] = struct{}{}
+		}
+
+		//Need to check if there is at least one pod in every zone to satisfy HA
+		if !state.SatisfyZoneAvailability(states.SchedulablePods, states) {
+			return 0, state.NewStatus(state.Unschedulable, ErrReasonNotEnoughPods)
+		}
+
+		zoneName, _, err := states.GetPodInfo(state.PodNameFromOrdinal(states.StatefulSetName, podID))
+		if err != nil {
+			return score, state.NewStatus(state.Error, ErrReasonNoResource)
+		}
+
+		currentReps := states.ZoneSpread[key][zoneName] //get #vreps on this zone
+		for otherZoneName := range zoneMap {            //compare with #vreps on other zones
+			if otherZoneName != zoneName {
+				otherReps := states.ZoneSpread[key][otherZoneName]
+				if skew = (currentReps + 1) - otherReps; skew < 0 {
+					skew = skew * int32(-1)
+				}
+
+				//logger.Infof("Current Zone %v with %d and Other Zone %v with %d causing skew %d", zoneName, currentReps, otherZoneName, otherReps, skew)
+				if skew > skewVal.MaxSkew { //score low
+					logger.Infof("Pod %d in zone %v will cause an uneven zone spread %v with other zone %v", podID, zoneName, states.ZoneSpread[key], otherZoneName)
+				}
+				score = score + uint64(skew)
+			}
+		}
+
+		score = math.MaxUint64 - score //lesser skews get higher score
+	}
+
+	return score, state.NewStatus(state.Success)
+}
+
+// ScoreExtensions of the Score plugin.
+func (pl *AvailabilityZonePriority) ScoreExtensions() state.ScoreExtensions {
+	return pl
+}
+
+// NormalizeScore invoked after scoring all pods.
+func (pl *AvailabilityZonePriority) NormalizeScore(ctx context.Context, states *state.State, scores state.PodScoreList) *state.Status {
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/evenpodspread/even_pod_spread.go b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/evenpodspread/even_pod_spread.go
new file mode 100644
index 000000000..edba30637
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/evenpodspread/even_pod_spread.go
@@ -0,0 +1,151 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package evenpodspread
+
+import (
+	"context"
+	"encoding/json"
+	"math"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/pkg/scheduler/factory"
+	state "knative.dev/eventing/pkg/scheduler/state"
+	"knative.dev/pkg/logging"
+)
+
+// EvenPodSpread is a filter or score plugin that picks/favors pods that create an equal spread of resources across pods
+type EvenPodSpread struct {
+}
+
+// Verify EvenPodSpread Implements FilterPlugin and ScorePlugin Interface
+var _ state.FilterPlugin = &EvenPodSpread{}
+var _ state.ScorePlugin = &EvenPodSpread{}
+
+// Name of the plugin
+const (
+	Name                   = state.EvenPodSpread
+	ErrReasonInvalidArg    = "invalid arguments"
+	ErrReasonUnschedulable = "pod will cause an uneven spread"
+)
+
+func init() {
+	factory.RegisterFP(Name, &EvenPodSpread{})
+	factory.RegisterSP(Name, &EvenPodSpread{})
+}
+
+// Name returns name of the plugin
+func (pl *EvenPodSpread) Name() string {
+	return Name
+}
+
+// Filter invoked at the filter extension point.
+func (pl *EvenPodSpread) Filter(ctx context.Context, args interface{}, states *state.State, key types.NamespacedName, podID int32) *state.Status {
+	logger := logging.FromContext(ctx).With("Filter", pl.Name())
+
+	spreadArgs, ok := args.(string)
+	if !ok {
+		logger.Errorf("Filter args %v for predicate %q are not valid", args, pl.Name())
+		return state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	skewVal := state.EvenPodSpreadArgs{}
+	decoder := json.NewDecoder(strings.NewReader(spreadArgs))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&skewVal); err != nil {
+		return state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	if states.Replicas > 0 { //need at least a pod to compute spread
+		currentReps := states.PodSpread[key][state.PodNameFromOrdinal(states.StatefulSetName, podID)] //get #vreps on this podID
+		var skew int32
+		for _, otherPodID := range states.SchedulablePods { //compare with #vreps on other pods
+			if otherPodID != podID {
+				otherReps := states.PodSpread[key][state.PodNameFromOrdinal(states.StatefulSetName, otherPodID)]
+
+				if otherReps == 0 && states.Free(otherPodID) <= 0 { //other pod fully occupied by other vpods - so ignore
+					continue
+				}
+				if skew = (currentReps + 1) - otherReps; skew < 0 {
+					skew = skew * int32(-1)
+				}
+
+				//logger.Infof("Current Pod %d with %d and Other Pod %d with %d causing skew %d", podID, currentReps, otherPodID, otherReps, skew)
+				if skew > skewVal.MaxSkew {
+					logger.Infof("Unschedulable! Pod %d will cause an uneven spread %v with other pod %v", podID, states.PodSpread[key], otherPodID)
+					return state.NewStatus(state.Unschedulable, ErrReasonUnschedulable)
+				}
+			}
+		}
+	}
+
+	return state.NewStatus(state.Success)
+}
+
+// Score invoked at the score extension point. The "score" returned in this function is higher for pods that create an even spread across pods.
+func (pl *EvenPodSpread) Score(ctx context.Context, args interface{}, states *state.State, feasiblePods []int32, key types.NamespacedName, podID int32) (uint64, *state.Status) {
+	logger := logging.FromContext(ctx).With("Score", pl.Name())
+	var score uint64 = 0
+
+	spreadArgs, ok := args.(string)
+	if !ok {
+		logger.Errorf("Scoring args %v for priority %q are not valid", args, pl.Name())
+		return 0, state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	skewVal := state.EvenPodSpreadArgs{}
+	decoder := json.NewDecoder(strings.NewReader(spreadArgs))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&skewVal); err != nil {
+		return 0, state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	if states.Replicas > 0 { //need at least a pod to compute spread
+		currentReps := states.PodSpread[key][state.PodNameFromOrdinal(states.StatefulSetName, podID)] //get #vreps on this podID
+		var skew int32
+		for _, otherPodID := range states.SchedulablePods { //compare with #vreps on other pods
+			if otherPodID != podID {
+				otherReps := states.PodSpread[key][state.PodNameFromOrdinal(states.StatefulSetName, otherPodID)]
+				if otherReps == 0 && states.Free(otherPodID) == 0 { //other pod fully occupied by other vpods - so ignore
+					continue
+				}
+				if skew = (currentReps + 1) - otherReps; skew < 0 {
+					skew = skew * int32(-1)
+				}
+
+				//logger.Infof("Current Pod %d with %d and Other Pod %d with %d causing skew %d", podID, currentReps, otherPodID, otherReps, skew)
+				if skew > skewVal.MaxSkew {
+					logger.Infof("Pod %d will cause an uneven spread %v with other pod %v", podID, states.PodSpread[key], otherPodID)
+				}
+				score = score + uint64(skew)
+			}
+		}
+		score = math.MaxUint64 - score //lesser skews get higher score
+	}
+
+	return score, state.NewStatus(state.Success)
+}
+
+// ScoreExtensions of the Score plugin.
+func (pl *EvenPodSpread) ScoreExtensions() state.ScoreExtensions {
+	return pl
+}
+
+// NormalizeScore invoked after scoring all pods.
+func (pl *EvenPodSpread) NormalizeScore(ctx context.Context, states *state.State, scores state.PodScoreList) *state.Status {
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/lowestordinalpriority/lowest_ordinal_priority.go b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/lowestordinalpriority/lowest_ordinal_priority.go
new file mode 100644
index 000000000..3068d5526
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/lowestordinalpriority/lowest_ordinal_priority.go
@@ -0,0 +1,61 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package lowestordinalpriority
+
+import (
+	"context"
+	"math"
+
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/pkg/scheduler/factory"
+	state "knative.dev/eventing/pkg/scheduler/state"
+)
+
+// LowestOrdinalPriority is a score plugin that favors pods that have a lower ordinal
+type LowestOrdinalPriority struct {
+}
+
+// Verify LowestOrdinalPriority Implements ScorePlugin Interface
+var _ state.ScorePlugin = &LowestOrdinalPriority{}
+
+// Name of the plugin
+const Name = state.LowestOrdinalPriority
+
+func init() {
+	factory.RegisterSP(Name, &LowestOrdinalPriority{})
+}
+
+// Name returns name of the plugin
+func (pl *LowestOrdinalPriority) Name() string {
+	return Name
+}
+
+// Score invoked at the score extension point. The "score" returned in this function is higher for pods with lower ordinal values.
+func (pl *LowestOrdinalPriority) Score(ctx context.Context, args interface{}, states *state.State, feasiblePods []int32, key types.NamespacedName, podID int32) (uint64, *state.Status) {
+	score := math.MaxUint64 - uint64(podID) //lower ordinals get higher score
+	return score, state.NewStatus(state.Success)
+}
+
+// ScoreExtensions of the Score plugin.
+func (pl *LowestOrdinalPriority) ScoreExtensions() state.ScoreExtensions {
+	return pl
+}
+
+// NormalizeScore invoked after scoring all pods.
+func (pl *LowestOrdinalPriority) NormalizeScore(ctx context.Context, states *state.State, scores state.PodScoreList) *state.Status {
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/podfitsresources/pod_fits_resources.go b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/podfitsresources/pod_fits_resources.go
new file mode 100644
index 000000000..e5fc2e619
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/podfitsresources/pod_fits_resources.go
@@ -0,0 +1,61 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package podfitsresources
+
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/pkg/scheduler/factory"
+	state "knative.dev/eventing/pkg/scheduler/state"
+	"knative.dev/pkg/logging"
+)
+
+// PodFitsResources is a plugin that filters pods that do not have sufficient free capacity for a vreplica to be placed on it
+type PodFitsResources struct {
+}
+
+// Verify PodFitsResources Implements FilterPlugin Interface
+var _ state.FilterPlugin = &PodFitsResources{}
+
+// Name of the plugin
+const Name = state.PodFitsResources
+
+const (
+	ErrReasonUnschedulable = "pod at full capacity"
+)
+
+func init() {
+	factory.RegisterFP(Name, &PodFitsResources{})
+}
+
+// Name returns name of the plugin
+func (pl *PodFitsResources) Name() string {
+	return Name
+}
+
+// Filter invoked at the filter extension point.
+func (pl *PodFitsResources) Filter(ctx context.Context, args interface{}, states *state.State, key types.NamespacedName, podID int32) *state.Status {
+	logger := logging.FromContext(ctx).With("Filter", pl.Name())
+
+	if len(states.FreeCap) == 0 || states.Free(podID) > 0 { //vpods with no placements or pods with positive free cap
+		return state.NewStatus(state.Success)
+	}
+
+	logger.Infof("Unschedulable! Pod %d has no free capacity %v", podID, states.FreeCap)
+	return state.NewStatus(state.Unschedulable, ErrReasonUnschedulable)
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/removewithavailabilitynodepriority/remove_with_availability_node_priority.go b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/removewithavailabilitynodepriority/remove_with_availability_node_priority.go
new file mode 100644
index 000000000..10f4e778c
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/removewithavailabilitynodepriority/remove_with_availability_node_priority.go
@@ -0,0 +1,113 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package removewithavailabilitynodepriority
+
+import (
+	"context"
+	"encoding/json"
+	"math"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/pkg/scheduler/factory"
+	state "knative.dev/eventing/pkg/scheduler/state"
+	"knative.dev/pkg/logging"
+)
+
+// RemoveWithAvailabilityNodePriority is a score plugin that favors pods that create an even spread of resources across nodes for HA
+type RemoveWithAvailabilityNodePriority struct {
+}
+
+// Verify RemoveWithAvailabilityNodePriority Implements ScorePlugin Interface
+var _ state.ScorePlugin = &RemoveWithAvailabilityNodePriority{}
+
+// Name of the plugin
+const Name = state.RemoveWithAvailabilityNodePriority
+
+const (
+	ErrReasonInvalidArg = "invalid arguments"
+	ErrReasonNoResource = "node does not exist"
+)
+
+func init() {
+	factory.RegisterSP(Name, &RemoveWithAvailabilityNodePriority{})
+}
+
+// Name returns name of the plugin
+func (pl *RemoveWithAvailabilityNodePriority) Name() string {
+	return Name
+}
+
+// Score invoked at the score extension point. The "score" returned in this function is higher for nodes that create an even spread across nodes.
+func (pl *RemoveWithAvailabilityNodePriority) Score(ctx context.Context, args interface{}, states *state.State, feasiblePods []int32, key types.NamespacedName, podID int32) (uint64, *state.Status) {
+	logger := logging.FromContext(ctx).With("Score", pl.Name())
+	var score uint64 = 0
+
+	spreadArgs, ok := args.(string)
+	if !ok {
+		logger.Errorf("Scoring args %v for priority %q are not valid", args, pl.Name())
+		return 0, state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	skewVal := state.AvailabilityNodePriorityArgs{}
+	decoder := json.NewDecoder(strings.NewReader(spreadArgs))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&skewVal); err != nil {
+		return 0, state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	if states.Replicas > 0 { //need at least a pod to compute spread
+		var skew int32
+		_, nodeName, err := states.GetPodInfo(state.PodNameFromOrdinal(states.StatefulSetName, podID))
+		if err != nil {
+			return score, state.NewStatus(state.Error, ErrReasonNoResource)
+		}
+
+		currentReps := states.NodeSpread[key][nodeName]   //get #vreps on this node
+		for otherNodeName := range states.NodeToZoneMap { //compare with #vreps on other pods
+			if otherNodeName != nodeName {
+				otherReps, ok := states.NodeSpread[key][otherNodeName]
+				if !ok {
+					continue //node does not exist in current placement, so move on
+				}
+				if skew = (currentReps - 1) - otherReps; skew < 0 {
+					skew = skew * int32(-1)
+				}
+
+				//logger.Infof("Current Node %v with %d and Other Node %v with %d causing skew %d", nodeName, currentReps, otherNodeName, otherReps, skew)
+				if skew > skewVal.MaxSkew { //score low
+					logger.Infof("Pod %d in node %v will cause an uneven node spread %v with other node %v", podID, nodeName, states.NodeSpread[key], otherNodeName)
+				}
+				score = score + uint64(skew)
+			}
+		}
+
+		score = math.MaxUint64 - score //lesser skews get higher score
+	}
+
+	return score, state.NewStatus(state.Success)
+}
+
+// ScoreExtensions of the Score plugin.
+func (pl *RemoveWithAvailabilityNodePriority) ScoreExtensions() state.ScoreExtensions {
+	return pl
+}
+
+// NormalizeScore invoked after scoring all pods.
+func (pl *RemoveWithAvailabilityNodePriority) NormalizeScore(ctx context.Context, states *state.State, scores state.PodScoreList) *state.Status {
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/removewithavailabilityzonepriority/remove_with_availability_zone_priority.go b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/removewithavailabilityzonepriority/remove_with_availability_zone_priority.go
new file mode 100644
index 000000000..9ac6d2774
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/removewithavailabilityzonepriority/remove_with_availability_zone_priority.go
@@ -0,0 +1,118 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package removewithavailabilityzonepriority
+
+import (
+	"context"
+	"encoding/json"
+	"math"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/pkg/scheduler/factory"
+	state "knative.dev/eventing/pkg/scheduler/state"
+	"knative.dev/pkg/logging"
+)
+
+// RemoveWithAvailabilityZonePriority is a score plugin that favors pods that create an even spread of resources across zones for HA
+type RemoveWithAvailabilityZonePriority struct {
+}
+
+// Verify RemoveWithAvailabilityZonePriority Implements ScorePlugin Interface
+var _ state.ScorePlugin = &RemoveWithAvailabilityZonePriority{}
+
+// Name of the plugin
+const Name = state.RemoveWithAvailabilityZonePriority
+
+const (
+	ErrReasonInvalidArg = "invalid arguments"
+	ErrReasonNoResource = "zone does not exist"
+)
+
+func init() {
+	factory.RegisterSP(Name, &RemoveWithAvailabilityZonePriority{})
+}
+
+// Name returns name of the plugin
+func (pl *RemoveWithAvailabilityZonePriority) Name() string {
+	return Name
+}
+
+// Score invoked at the score extension point. The "score" returned in this function is higher for zones that create an even spread across zones.
+func (pl *RemoveWithAvailabilityZonePriority) Score(ctx context.Context, args interface{}, states *state.State, feasiblePods []int32, key types.NamespacedName, podID int32) (uint64, *state.Status) {
+	logger := logging.FromContext(ctx).With("Score", pl.Name())
+	var score uint64 = 0
+
+	spreadArgs, ok := args.(string)
+	if !ok {
+		logger.Errorf("Scoring args %v for priority %q are not valid", args, pl.Name())
+		return 0, state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	skewVal := state.AvailabilityZonePriorityArgs{}
+	decoder := json.NewDecoder(strings.NewReader(spreadArgs))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&skewVal); err != nil {
+		return 0, state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	if states.Replicas > 0 { //need at least a pod to compute spread
+		var skew int32
+		zoneMap := make(map[string]struct{})
+		for _, zoneName := range states.NodeToZoneMap {
+			zoneMap[zoneName] = struct{}{}
+		}
+
+		zoneName, _, err := states.GetPodInfo(state.PodNameFromOrdinal(states.StatefulSetName, podID))
+		if err != nil {
+			return score, state.NewStatus(state.Error, ErrReasonNoResource)
+		}
+
+		currentReps := states.ZoneSpread[key][zoneName] //get #vreps on this zone
+		for otherZoneName := range zoneMap {            //compare with #vreps on other pods
+			if otherZoneName != zoneName {
+				otherReps, ok := states.ZoneSpread[key][otherZoneName]
+				if !ok {
+					continue //zone does not exist in current placement, so move on
+				}
+				if skew = (currentReps - 1) - otherReps; skew < 0 {
+					skew = skew * int32(-1)
+				}
+
+				//logger.Infof("Current Zone %v with %d and Other Zone %v with %d causing skew %d", zoneName, currentReps, otherZoneName, otherReps, skew)
+				if skew > skewVal.MaxSkew { //score low
+					logger.Infof("Pod %d in zone %v will cause an uneven zone spread %v with other zone %v", podID, zoneName, states.ZoneSpread[key], otherZoneName)
+				}
+				score = score + uint64(skew)
+			}
+		}
+
+		score = math.MaxUint64 - score //lesser skews get higher score
+	}
+
+	return score, state.NewStatus(state.Success)
+}
+
+// ScoreExtensions of the Score plugin.
+func (pl *RemoveWithAvailabilityZonePriority) ScoreExtensions() state.ScoreExtensions {
+	return pl
+}
+
+// NormalizeScore invoked after scoring all pods.
+func (pl *RemoveWithAvailabilityZonePriority) NormalizeScore(ctx context.Context, states *state.State, scores state.PodScoreList) *state.Status {
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/removewithevenpodspreadpriority/remove_with_even_pod_spread_priority.go b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/removewithevenpodspreadpriority/remove_with_even_pod_spread_priority.go
new file mode 100644
index 000000000..8c7692d66
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/removewithevenpodspreadpriority/remove_with_even_pod_spread_priority.go
@@ -0,0 +1,106 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package removewithevenpodspreadpriority
+
+import (
+	"context"
+	"encoding/json"
+	"math"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/pkg/scheduler/factory"
+	state "knative.dev/eventing/pkg/scheduler/state"
+	"knative.dev/pkg/logging"
+)
+
+// RemoveWithEvenPodSpreadPriority is a filter plugin that eliminates pods that do not create an equal spread of resources across pods
+type RemoveWithEvenPodSpreadPriority struct {
+}
+
+// Verify RemoveWithEvenPodSpreadPriority Implements FilterPlugin Interface
+var _ state.ScorePlugin = &RemoveWithEvenPodSpreadPriority{}
+
+// Name of the plugin
+const (
+	Name                   = state.RemoveWithEvenPodSpreadPriority
+	ErrReasonInvalidArg    = "invalid arguments"
+	ErrReasonUnschedulable = "pod will cause an uneven spread"
+)
+
+func init() {
+	factory.RegisterSP(Name, &RemoveWithEvenPodSpreadPriority{})
+}
+
+// Name returns name of the plugin
+func (pl *RemoveWithEvenPodSpreadPriority) Name() string {
+	return Name
+}
+
+// Score invoked at the score extension point. The "score" returned in this function is higher for pods that create an even spread across pods.
+func (pl *RemoveWithEvenPodSpreadPriority) Score(ctx context.Context, args interface{}, states *state.State, feasiblePods []int32, key types.NamespacedName, podID int32) (uint64, *state.Status) {
+	logger := logging.FromContext(ctx).With("Score", pl.Name())
+	var score uint64 = 0
+
+	spreadArgs, ok := args.(string)
+	if !ok {
+		logger.Errorf("Scoring args %v for priority %q are not valid", args, pl.Name())
+		return 0, state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	skewVal := state.EvenPodSpreadArgs{}
+	decoder := json.NewDecoder(strings.NewReader(spreadArgs))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&skewVal); err != nil {
+		return 0, state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	if states.Replicas > 0 { //need at least a pod to compute spread
+		currentReps := states.PodSpread[key][state.PodNameFromOrdinal(states.StatefulSetName, podID)] //get #vreps on this podID
+		var skew int32
+		for _, otherPodID := range states.SchedulablePods { //compare with #vreps on other pods
+			if otherPodID != podID {
+				otherReps, ok := states.PodSpread[key][state.PodNameFromOrdinal(states.StatefulSetName, otherPodID)]
+				if !ok {
+					continue //pod does not exist in current placement, so move on
+				}
+				if skew = (currentReps - 1) - otherReps; skew < 0 {
+					skew = skew * int32(-1)
+				}
+
+				//logger.Infof("Current Pod %v with %d and Other Pod %v with %d causing skew %d", podID, currentReps, otherPodID, otherReps, skew)
+				if skew > skewVal.MaxSkew { //score low
+					logger.Infof("Pod %d will cause an uneven spread %v with other pod %v", podID, states.PodSpread[key], otherPodID)
+				}
+				score = score + uint64(skew)
+			}
+		}
+		score = math.MaxUint64 - score //lesser skews get higher score
+	}
+
+	return score, state.NewStatus(state.Success)
+}
+
+// ScoreExtensions of the Score plugin.
+func (pl *RemoveWithEvenPodSpreadPriority) ScoreExtensions() state.ScoreExtensions {
+	return pl
+}
+
+// NormalizeScore invoked after scoring all pods.
+func (pl *RemoveWithEvenPodSpreadPriority) NormalizeScore(ctx context.Context, states *state.State, scores state.PodScoreList) *state.Status {
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/removewithhighestordinalpriority/remove_with_highest_ordinal_priority.go b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/removewithhighestordinalpriority/remove_with_highest_ordinal_priority.go
new file mode 100644
index 000000000..9a390977f
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/plugins/core/removewithhighestordinalpriority/remove_with_highest_ordinal_priority.go
@@ -0,0 +1,60 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package removewithhighestordinalpriority
+
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/pkg/scheduler/factory"
+	state "knative.dev/eventing/pkg/scheduler/state"
+)
+
+// RemoveWithHighestOrdinalPriority is a score plugin that favors pods that have a higher ordinal
+type RemoveWithHighestOrdinalPriority struct {
+}
+
+// Verify RemoveWithHighestOrdinalPriority Implements ScorePlugin Interface
+var _ state.ScorePlugin = &RemoveWithHighestOrdinalPriority{}
+
+// Name of the plugin
+const Name = state.RemoveWithHighestOrdinalPriority
+
+func init() {
+	factory.RegisterSP(Name, &RemoveWithHighestOrdinalPriority{})
+}
+
+// Name returns name of the plugin
+func (pl *RemoveWithHighestOrdinalPriority) Name() string {
+	return Name
+}
+
+// Score invoked at the score extension point. The "score" returned in this function is higher for pods with higher ordinal values.
+func (pl *RemoveWithHighestOrdinalPriority) Score(ctx context.Context, args interface{}, states *state.State, feasiblePods []int32, key types.NamespacedName, podID int32) (uint64, *state.Status) {
+	score := uint64(podID) //higher ordinals get higher score
+	return score, state.NewStatus(state.Success)
+}
+
+// ScoreExtensions of the Score plugin.
+func (pl *RemoveWithHighestOrdinalPriority) ScoreExtensions() state.ScoreExtensions {
+	return pl
+}
+
+// NormalizeScore invoked after scoring all pods.
+func (pl *RemoveWithHighestOrdinalPriority) NormalizeScore(ctx context.Context, states *state.State, scores state.PodScoreList) *state.Status {
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/plugins/kafka/nomaxresourcecount/no_max_resource_count.go b/vendor/knative.dev/eventing/pkg/scheduler/plugins/kafka/nomaxresourcecount/no_max_resource_count.go
new file mode 100644
index 000000000..301d29293
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/plugins/kafka/nomaxresourcecount/no_max_resource_count.go
@@ -0,0 +1,78 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package nomaxresourcecount
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/pkg/scheduler/factory"
+	state "knative.dev/eventing/pkg/scheduler/state"
+	"knative.dev/pkg/logging"
+)
+
+// NoMaxResourceCount plugin filters pods that cause total pods with placements to exceed total partitioncount.
+type NoMaxResourceCount struct {
+}
+
+// Verify NoMaxResourceCount Implements FilterPlugin Interface
+var _ state.FilterPlugin = &NoMaxResourceCount{}
+
+// Name of the plugin
+const Name = state.NoMaxResourceCount
+
+const (
+	ErrReasonInvalidArg    = "invalid arguments"
+	ErrReasonUnschedulable = "pod increases total # of pods beyond partition count"
+)
+
+func init() {
+	factory.RegisterFP(Name, &NoMaxResourceCount{})
+}
+
+// Name returns name of the plugin
+func (pl *NoMaxResourceCount) Name() string {
+	return Name
+}
+
+// Filter invoked at the filter extension point.
+func (pl *NoMaxResourceCount) Filter(ctx context.Context, args interface{}, states *state.State, key types.NamespacedName, podID int32) *state.Status {
+	logger := logging.FromContext(ctx).With("Filter", pl.Name())
+
+	resourceCountArgs, ok := args.(string)
+	if !ok {
+		logger.Errorf("Filter args %v for predicate %q are not valid", args, pl.Name())
+		return state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	resVal := state.NoMaxResourceCountArgs{}
+	decoder := json.NewDecoder(strings.NewReader(resourceCountArgs))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&resVal); err != nil {
+		return state.NewStatus(state.Unschedulable, ErrReasonInvalidArg)
+	}
+
+	podName := state.PodNameFromOrdinal(states.StatefulSetName, podID)
+	if _, ok := states.PodSpread[key][podName]; !ok && ((len(states.PodSpread[key]) + 1) > resVal.NumPartitions) { //pod not in vrep's partition map and counting this new pod towards total pod count
+		logger.Infof("Unschedulable! Pod %d filtered due to total pod count %v exceeding partition count", podID, len(states.PodSpread[key])+1)
+		return state.NewStatus(state.Unschedulable, ErrReasonUnschedulable)
+	}
+
+	return state.NewStatus(state.Success)
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/state/helpers.go b/vendor/knative.dev/eventing/pkg/scheduler/state/helpers.go
new file mode 100644
index 000000000..56ca82b8d
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/state/helpers.go
@@ -0,0 +1,78 @@
+/*
+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 state
+
+import (
+	"math"
+	"strconv"
+	"strings"
+	"time"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"knative.dev/eventing/pkg/scheduler"
+)
+
+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)
+}
+
+// Get retrieves the VPod from the vpods lister for a given namespace and name.
+func GetVPod(key types.NamespacedName, vpods []scheduler.VPod) scheduler.VPod {
+	for _, vpod := range vpods {
+		if vpod.GetKey().Name == key.Name && vpod.GetKey().Namespace == key.Namespace {
+			return vpod
+		}
+	}
+	return nil
+}
+
+func SatisfyZoneAvailability(feasiblePods []int32, states *State) bool {
+	zoneMap := make(map[string]struct{})
+	var zoneName string
+	var err error
+	for _, podID := range feasiblePods {
+		wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
+			zoneName, _, err = states.GetPodInfo(PodNameFromOrdinal(states.StatefulSetName, podID))
+			return err == nil, nil
+		})
+		zoneMap[zoneName] = struct{}{}
+	}
+	return len(zoneMap) == int(states.NumZones)
+}
+
+func SatisfyNodeAvailability(feasiblePods []int32, states *State) bool {
+	nodeMap := make(map[string]struct{})
+	var nodeName string
+	var err error
+	for _, podID := range feasiblePods {
+		wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
+			_, nodeName, err = states.GetPodInfo(PodNameFromOrdinal(states.StatefulSetName, podID))
+			return err == nil, nil
+		})
+		nodeMap[nodeName] = struct{}{}
+	}
+	return len(nodeMap) == int(states.NumNodes)
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/state/interface.go b/vendor/knative.dev/eventing/pkg/scheduler/state/interface.go
new file mode 100644
index 000000000..664fb3e4f
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/state/interface.go
@@ -0,0 +1,209 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package state
+
+import (
+	"context"
+	"errors"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/types"
+)
+
+const (
+	PodFitsResources                   = "PodFitsResources"
+	NoMaxResourceCount                 = "NoMaxResourceCount"
+	EvenPodSpread                      = "EvenPodSpread"
+	AvailabilityNodePriority           = "AvailabilityNodePriority"
+	AvailabilityZonePriority           = "AvailabilityZonePriority"
+	LowestOrdinalPriority              = "LowestOrdinalPriority"
+	RemoveWithEvenPodSpreadPriority    = "RemoveWithEvenPodSpreadPriority"
+	RemoveWithAvailabilityNodePriority = "RemoveWithAvailabilityNodePriority"
+	RemoveWithAvailabilityZonePriority = "RemoveWithAvailabilityZonePriority"
+	RemoveWithHighestOrdinalPriority   = "RemoveWithHighestOrdinalPriority"
+)
+
+// Plugin is the parent type for all the scheduling framework plugins.
+type Plugin interface {
+	Name() string
+}
+
+type FilterPlugin interface {
+	Plugin
+	// Filter is called by the scheduler.
+	// All FilterPlugins should return "Success" to declare that
+	// the given pod fits the vreplica.
+	Filter(ctx context.Context, args interface{}, state *State, key types.NamespacedName, podID int32) *Status
+}
+
+// ScoreExtensions is an interface for Score extended functionality.
+type ScoreExtensions interface {
+	// NormalizeScore is called for all pod scores produced by the same plugin's "Score"
+	// method. A successful run of NormalizeScore will update the scores list and return
+	// a success status.
+	NormalizeScore(ctx context.Context, state *State, scores PodScoreList) *Status
+}
+
+type ScorePlugin interface {
+	Plugin
+	// Score is called by the scheduler.
+	// All ScorePlugins should return "Success" unless the args are invalid.
+	Score(ctx context.Context, args interface{}, state *State, feasiblePods []int32, key types.NamespacedName, podID int32) (uint64, *Status)
+
+	// ScoreExtensions returns a ScoreExtensions interface if it implements one, or nil if does not
+	ScoreExtensions() ScoreExtensions
+}
+
+// NoMaxResourceCountArgs holds arguments used to configure the NoMaxResourceCount plugin.
+type NoMaxResourceCountArgs struct {
+	NumPartitions int
+}
+
+// EvenPodSpreadArgs holds arguments used to configure the EvenPodSpread plugin.
+type EvenPodSpreadArgs struct {
+	MaxSkew int32
+}
+
+// AvailabilityZonePriorityArgs holds arguments used to configure the AvailabilityZonePriority plugin.
+type AvailabilityZonePriorityArgs struct {
+	MaxSkew int32
+}
+
+// AvailabilityNodePriorityArgs holds arguments used to configure the AvailabilityNodePriority plugin.
+type AvailabilityNodePriorityArgs struct {
+	MaxSkew int32
+}
+
+// Code is the Status code/type which is returned from plugins.
+type Code int
+
+// These are predefined codes used in a Status.
+const (
+	// Success means that plugin ran correctly and found pod schedulable.
+	Success Code = iota
+	// Unschedulable is used when a plugin finds a pod unschedulable due to not satisying the predicate.
+	Unschedulable
+	// Error is used for internal plugin errors, unexpected input, etc.
+	Error
+)
+
+// Status indicates the result of running a plugin.
+type Status struct {
+	code    Code
+	reasons []string
+	err     error
+}
+
+// Code returns code of the Status.
+func (s *Status) Code() Code {
+	if s == nil {
+		return Success
+	}
+	return s.code
+}
+
+// Message returns a concatenated message on reasons of the Status.
+func (s *Status) Message() string {
+	if s == nil {
+		return ""
+	}
+	return strings.Join(s.reasons, ", ")
+}
+
+// NewStatus makes a Status out of the given arguments and returns its pointer.
+func NewStatus(code Code, reasons ...string) *Status {
+	s := &Status{
+		code:    code,
+		reasons: reasons,
+	}
+	if code == Error {
+		s.err = errors.New(s.Message())
+	}
+	return s
+}
+
+// AsStatus wraps an error in a Status.
+func AsStatus(err error) *Status {
+	return &Status{
+		code:    Error,
+		reasons: []string{err.Error()},
+		err:     err,
+	}
+}
+
+// AsError returns nil if the status is a success; otherwise returns an "error" object
+// with a concatenated message on reasons of the Status.
+func (s *Status) AsError() error {
+	if s.IsSuccess() {
+		return nil
+	}
+	if s.err != nil {
+		return s.err
+	}
+	return errors.New(s.Message())
+}
+
+// IsSuccess returns true if and only if "Status" is nil or Code is "Success".
+func (s *Status) IsSuccess() bool {
+	return s.Code() == Success
+}
+
+// IsError returns true if and only if "Status" is "Error".
+func (s *Status) IsError() bool {
+	return s.Code() == Error
+}
+
+// IsUnschedulable returns true if "Status" is Unschedulable
+func (s *Status) IsUnschedulable() bool {
+	return s.Code() == Unschedulable
+}
+
+type PodScore struct {
+	ID    int32
+	Score uint64
+}
+
+type PodScoreList []PodScore
+
+// PluginToPodScores declares a map from plugin name to its PodScoreList.
+type PluginToPodScores map[string]PodScoreList
+
+// PluginToStatus maps plugin name to status. Currently used to identify which Filter plugin
+// returned which status.
+type PluginToStatus map[string]*Status
+
+// Merge merges the statuses in the map into one. The resulting status code have the following
+// precedence: Error, Unschedulable, Success
+func (p PluginToStatus) Merge() *Status {
+	if len(p) == 0 {
+		return nil
+	}
+
+	finalStatus := NewStatus(Success)
+	for _, s := range p {
+		if s.Code() == Error {
+			finalStatus.err = s.AsError()
+		}
+		if s.Code() > finalStatus.code {
+			finalStatus.code = s.Code()
+		}
+
+		finalStatus.reasons = append(finalStatus.reasons, s.reasons...)
+	}
+
+	return finalStatus
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/state/state.go b/vendor/knative.dev/eventing/pkg/scheduler/state/state.go
new file mode 100644
index 000000000..f3b7f12fb
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/state/state.go
@@ -0,0 +1,392 @@
+/*
+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 state
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	"go.uber.org/zap"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	clientappsv1 "k8s.io/client-go/kubernetes/typed/apps/v1"
+	corev1 "k8s.io/client-go/listers/core/v1"
+
+	scheduler "knative.dev/eventing/pkg/scheduler"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/logging"
+)
+
+type StateAccessor interface {
+	// State returns the current state (snapshot) about placed vpods
+	// Take into account reserved vreplicas and update `reserved` to reflect
+	// the current state.
+	State(reserved map[types.NamespacedName]map[string]int32) (*State, error)
+}
+
+// state provides information about the current scheduling of all vpods
+// It is used by for the scheduler and the autoscaler
+type State struct {
+	// free tracks the free capacity of each pod.
+	FreeCap []int32
+
+	// schedulable pods tracks the pods that aren't being evicted.
+	SchedulablePods []int32
+
+	// LastOrdinal is the ordinal index corresponding to the last statefulset replica
+	// with placed vpods.
+	LastOrdinal int32
+
+	// Pod capacity.
+	Capacity int32
+
+	// Replicas is the (cached) number of statefulset replicas.
+	Replicas int32
+
+	// Number of available zones in cluster
+	NumZones int32
+
+	// Number of available nodes in cluster
+	NumNodes int32
+
+	// Scheduling policy type for placing vreplicas on pods
+	SchedulerPolicy scheduler.SchedulerPolicyType
+
+	// Scheduling policy plugin for placing vreplicas on pods
+	SchedPolicy *scheduler.SchedulerPolicy
+
+	// De-scheduling policy plugin for removing vreplicas from pods
+	DeschedPolicy *scheduler.SchedulerPolicy
+
+	// Mapping node names of nodes currently in cluster to their zone info
+	NodeToZoneMap map[string]string
+
+	StatefulSetName string
+
+	PodLister corev1.PodNamespaceLister
+
+	// Stores for each vpod, a map of podname to number of vreplicas placed on that pod currently
+	PodSpread map[types.NamespacedName]map[string]int32
+
+	// Stores for each vpod, a map of nodename to total number of vreplicas placed on all pods running on that node currently
+	NodeSpread map[types.NamespacedName]map[string]int32
+
+	// Stores for each vpod, a map of zonename to total number of vreplicas placed on all pods located in that zone currently
+	ZoneSpread map[types.NamespacedName]map[string]int32
+}
+
+// Free safely returns the free capacity at the given ordinal
+func (s *State) Free(ordinal int32) int32 {
+	if int32(len(s.FreeCap)) <= ordinal {
+		return s.Capacity
+	}
+	return s.FreeCap[ordinal]
+}
+
+// SetFree safely sets the free capacity at the given ordinal
+func (s *State) SetFree(ordinal int32, value int32) {
+	s.FreeCap = grow(s.FreeCap, ordinal, s.Capacity)
+	s.FreeCap[int(ordinal)] = value
+}
+
+// freeCapacity returns the number of vreplicas that can be used,
+// up to the last ordinal
+func (s *State) FreeCapacity() int32 {
+	t := int32(0)
+	for _, i := range s.SchedulablePods {
+		t += s.FreeCap[i]
+	}
+	return t
+}
+
+func (s *State) GetPodInfo(podName string) (zoneName string, nodeName string, err error) {
+	pod, err := s.PodLister.Get(podName)
+	if err != nil {
+		return zoneName, nodeName, err
+	}
+
+	nodeName = pod.Spec.NodeName
+	zoneName, ok := s.NodeToZoneMap[nodeName]
+	if !ok {
+		return zoneName, nodeName, errors.New("could not find zone")
+	}
+	return zoneName, nodeName, nil
+}
+
+// stateBuilder reconstruct the state from scratch, by listing vpods
+type stateBuilder struct {
+	ctx               context.Context
+	logger            *zap.SugaredLogger
+	vpodLister        scheduler.VPodLister
+	capacity          int32
+	schedulerPolicy   scheduler.SchedulerPolicyType
+	nodeLister        corev1.NodeLister
+	statefulSetClient clientappsv1.StatefulSetInterface
+	statefulSetName   string
+	podLister         corev1.PodNamespaceLister
+	schedPolicy       *scheduler.SchedulerPolicy
+	deschedPolicy     *scheduler.SchedulerPolicy
+}
+
+// NewStateBuilder returns a StateAccessor recreating the state from scratch each time it is requested
+func NewStateBuilder(ctx context.Context, namespace, sfsname string, lister scheduler.VPodLister, podCapacity int32, schedulerPolicy scheduler.SchedulerPolicyType, schedPolicy *scheduler.SchedulerPolicy, deschedPolicy *scheduler.SchedulerPolicy, podlister corev1.PodNamespaceLister, nodeLister corev1.NodeLister) StateAccessor {
+
+	return &stateBuilder{
+		ctx:               ctx,
+		logger:            logging.FromContext(ctx),
+		vpodLister:        lister,
+		capacity:          podCapacity,
+		schedulerPolicy:   schedulerPolicy,
+		nodeLister:        nodeLister,
+		statefulSetClient: kubeclient.Get(ctx).AppsV1().StatefulSets(namespace),
+		statefulSetName:   sfsname,
+		podLister:         podlister,
+		schedPolicy:       schedPolicy,
+		deschedPolicy:     deschedPolicy,
+	}
+}
+
+func (s *stateBuilder) State(reserved map[types.NamespacedName]map[string]int32) (*State, error) {
+	vpods, err := s.vpodLister()
+	if err != nil {
+		return nil, err
+	}
+
+	scale, err := s.statefulSetClient.GetScale(s.ctx, s.statefulSetName, metav1.GetOptions{})
+	if err != nil {
+		s.logger.Infow("failed to get statefulset", zap.Error(err))
+		return nil, err
+	}
+
+	free := make([]int32, 0)
+	schedulablePods := make([]int32, 0)
+	last := int32(-1)
+
+	// keep track of (vpod key, podname) pairs with existing placements
+	withPlacement := make(map[types.NamespacedName]map[string]bool)
+
+	podSpread := make(map[types.NamespacedName]map[string]int32)
+	nodeSpread := make(map[types.NamespacedName]map[string]int32)
+	zoneSpread := make(map[types.NamespacedName]map[string]int32)
+
+	noexec := &v1.Taint{
+		Key:    "node.kubernetes.io/unreachable",
+		Effect: v1.TaintEffectNoExecute,
+	}
+
+	nosched := &v1.Taint{
+		Key:    "node.kubernetes.io/unreachable",
+		Effect: v1.TaintEffectNoSchedule,
+	}
+
+	//Build the node to zone map
+	nodes, err := s.nodeLister.List(labels.Everything())
+	if err != nil {
+		return nil, err
+	}
+
+	nodeToZoneMap := make(map[string]string)
+	zoneMap := make(map[string]struct{})
+	for i := 0; i < len(nodes); i++ {
+		node := nodes[i]
+		if node.Spec.Unschedulable {
+			continue //ignore node that is currently unschedulable
+		}
+
+		taints := node.Spec.Taints
+		if contains(taints, noexec) || contains(taints, nosched) {
+			continue //ignore node that is currently unschedulable
+		}
+
+		zoneName, ok := node.GetLabels()[scheduler.ZoneLabel]
+		if !ok {
+			continue //ignore node that doesn't have zone info (maybe a test setup or control node)
+		}
+
+		nodeToZoneMap[node.Name] = zoneName
+		zoneMap[zoneName] = struct{}{}
+	}
+
+	for podId := int32(0); podId < scale.Spec.Replicas && s.podLister != nil; podId++ {
+		var pod *v1.Pod
+		wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
+			pod, err = s.podLister.Get(PodNameFromOrdinal(s.statefulSetName, podId))
+			return err == nil, nil
+		})
+
+		if pod != nil {
+			var unschedulable bool
+			annotVal, ok := pod.ObjectMeta.Annotations[scheduler.PodAnnotationKey] //Pod is marked for eviction - CANNOT SCHEDULE VREPS on this pod
+			if ok {
+				unschedulable, _ = strconv.ParseBool(annotVal)
+			}
+
+			nodeName := pod.Spec.NodeName
+			node, err := s.nodeLister.Get(nodeName) //Node could be marked as Unschedulable - CANNOT SCHEDULE VREPS on a pod running on this node
+			if err != nil {
+				return nil, err
+			}
+
+			_, okZone := node.GetLabels()[scheduler.ZoneLabel] //Node has no zone info (maybe zone is down or a control node) - CANNOT SCHEDULE VREPS on a pod running on this node
+
+			taints := node.Spec.Taints
+
+			if (!ok || !unschedulable) && !node.Spec.Unschedulable && okZone && !contains(taints, noexec) && !contains(taints, nosched) { //Pod has no annotation or not annotated as unschedulable and not on an unschedulable node, so add to feasible
+				schedulablePods = append(schedulablePods, podId)
+			}
+		}
+	}
+
+	// Getting current state from existing placements for all vpods
+	for _, vpod := range vpods {
+		ps := vpod.GetPlacements()
+
+		withPlacement[vpod.GetKey()] = make(map[string]bool)
+		podSpread[vpod.GetKey()] = make(map[string]int32)
+		nodeSpread[vpod.GetKey()] = make(map[string]int32)
+		zoneSpread[vpod.GetKey()] = make(map[string]int32)
+
+		for i := 0; i < len(ps); i++ {
+			podName := ps[i].PodName
+			vreplicas := ps[i].VReplicas
+
+			// Account for reserved vreplicas
+			vreplicas = withReserved(vpod.GetKey(), podName, vreplicas, reserved)
+
+			free, last = s.updateFreeCapacity(free, last, podName, vreplicas)
+
+			withPlacement[vpod.GetKey()][podName] = true
+
+			var pod *v1.Pod
+			wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
+				pod, err = s.podLister.Get(podName)
+				return err == nil, nil
+			})
+
+			if pod != nil {
+				nodeName := pod.Spec.NodeName       //node name for this pod
+				zoneName := nodeToZoneMap[nodeName] //zone name for this pod
+				podSpread[vpod.GetKey()][podName] = podSpread[vpod.GetKey()][podName] + vreplicas
+				nodeSpread[vpod.GetKey()][nodeName] = nodeSpread[vpod.GetKey()][nodeName] + vreplicas
+				zoneSpread[vpod.GetKey()][zoneName] = zoneSpread[vpod.GetKey()][zoneName] + vreplicas
+			}
+		}
+	}
+
+	// Account for reserved vreplicas with no prior placements
+	for key, ps := range reserved {
+		for podName, rvreplicas := range ps {
+			if wp, ok := withPlacement[key]; ok {
+				if _, ok := wp[podName]; ok {
+					// already accounted for
+					continue
+				}
+
+				var pod *v1.Pod
+				wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
+					pod, err = s.podLister.Get(podName)
+					return err == nil, nil
+				})
+
+				if pod != nil {
+					nodeName := pod.Spec.NodeName       //node name for this pod
+					zoneName := nodeToZoneMap[nodeName] //zone name for this pod
+					podSpread[key][podName] = podSpread[key][podName] + rvreplicas
+					nodeSpread[key][nodeName] = nodeSpread[key][nodeName] + rvreplicas
+					zoneSpread[key][zoneName] = zoneSpread[key][zoneName] + rvreplicas
+				}
+			}
+
+			free, last = s.updateFreeCapacity(free, last, podName, rvreplicas)
+		}
+	}
+
+	s.logger.Infow("cluster state info", zap.String("NumPods", fmt.Sprint(scale.Spec.Replicas)), zap.String("NumZones", fmt.Sprint(len(zoneMap))), zap.String("NumNodes", fmt.Sprint(len(nodeToZoneMap))), zap.String("Schedulable", fmt.Sprint(schedulablePods)))
+	return &State{FreeCap: free, SchedulablePods: schedulablePods, LastOrdinal: last, Capacity: s.capacity, Replicas: scale.Spec.Replicas, NumZones: int32(len(zoneMap)), NumNodes: int32(len(nodeToZoneMap)),
+		SchedulerPolicy: s.schedulerPolicy, SchedPolicy: s.schedPolicy, DeschedPolicy: s.deschedPolicy, NodeToZoneMap: nodeToZoneMap, StatefulSetName: s.statefulSetName, PodLister: s.podLister,
+		PodSpread: podSpread, NodeSpread: nodeSpread, ZoneSpread: zoneSpread}, nil
+}
+
+func (s *stateBuilder) updateFreeCapacity(free []int32, last int32, podName string, vreplicas int32) ([]int32, int32) {
+	ordinal := OrdinalFromPodName(podName)
+	free = grow(free, ordinal, s.capacity)
+
+	free[ordinal] -= vreplicas
+
+	// Assert the pod is not overcommitted
+	if free[ordinal] < 0 {
+		// This should not happen anymore. Log as an error but do not interrupt the current scheduling.
+		s.logger.Errorw("pod is overcommitted", zap.String("podName", podName), zap.Int32("free", free[ordinal]))
+	}
+
+	if ordinal > last && free[ordinal] != s.capacity {
+		last = ordinal
+	}
+
+	return free, last
+}
+
+func grow(slice []int32, ordinal int32, def int32) []int32 {
+	l := int32(len(slice))
+	diff := ordinal - l + 1
+
+	if diff <= 0 {
+		return slice
+	}
+
+	for i := int32(0); i < diff; i++ {
+		slice = append(slice, def)
+	}
+	return slice
+}
+
+func withReserved(key types.NamespacedName, podName string, committed int32, reserved map[types.NamespacedName]map[string]int32) int32 {
+	if reserved != nil {
+		if rps, ok := reserved[key]; ok {
+			if rvreplicas, ok := rps[podName]; ok {
+				if committed == rvreplicas {
+					// new placement has been committed.
+					delete(rps, podName)
+					if len(rps) == 0 {
+						delete(reserved, key)
+					}
+				} else {
+					// new placement hasn't been committed yet. Adjust locally
+					// needed for descheduling vreps using policies
+					return rvreplicas
+				}
+			}
+		}
+	}
+	return committed
+}
+
+func contains(taints []v1.Taint, taint *v1.Taint) bool {
+	for _, v := range taints {
+		if v.MatchTaint(taint) {
+			return true
+		}
+	}
+
+	return false
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/statefulset/autoscaler.go b/vendor/knative.dev/eventing/pkg/scheduler/statefulset/autoscaler.go
new file mode 100644
index 000000000..fb5da1c9c
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/statefulset/autoscaler.go
@@ -0,0 +1,284 @@
+/*
+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"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	clientappsv1 "k8s.io/client-go/kubernetes/typed/apps/v1"
+
+	"knative.dev/eventing/pkg/scheduler"
+	st "knative.dev/eventing/pkg/scheduler/state"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/logging"
+)
+
+type Autoscaler interface {
+	// Start runs the autoscaler until cancelled.
+	Start(ctx context.Context)
+
+	// Autoscale is used to immediately trigger the autoscaler with the hint
+	// that pending number of vreplicas couldn't be scheduled.
+	Autoscale(ctx context.Context, attemptScaleDown bool, pending int32)
+}
+
+type autoscaler struct {
+	statefulSetClient clientappsv1.StatefulSetInterface
+	statefulSetName   string
+	vpodLister        scheduler.VPodLister
+	logger            *zap.SugaredLogger
+	stateAccessor     st.StateAccessor
+	trigger           chan int32
+	evictor           scheduler.Evictor
+
+	// capacity is the total number of virtual replicas available per pod.
+	capacity int32
+
+	// refreshPeriod is how often the autoscaler tries to scale down the statefulset
+	refreshPeriod time.Duration
+	lock          sync.Locker
+}
+
+func NewAutoscaler(ctx context.Context,
+	namespace, name string,
+	lister scheduler.VPodLister,
+	stateAccessor st.StateAccessor,
+	evictor scheduler.Evictor,
+	refreshPeriod time.Duration,
+	capacity int32) Autoscaler {
+
+	return &autoscaler{
+		logger:            logging.FromContext(ctx),
+		statefulSetClient: kubeclient.Get(ctx).AppsV1().StatefulSets(namespace),
+		statefulSetName:   name,
+		vpodLister:        lister,
+		stateAccessor:     stateAccessor,
+		evictor:           evictor,
+		trigger:           make(chan int32, 1),
+		capacity:          capacity,
+		refreshPeriod:     refreshPeriod,
+		lock:              new(sync.Mutex),
+	}
+}
+
+func (a *autoscaler) Start(ctx context.Context) {
+	attemptScaleDown := false
+	pending := int32(0)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(a.refreshPeriod):
+			attemptScaleDown = true
+		case pending = <-a.trigger:
+			attemptScaleDown = false
+		}
+
+		// Retry a few times, just so that we don't have to wait for the next beat when
+		// a transient error occurs
+		a.syncAutoscale(ctx, attemptScaleDown, pending)
+		pending = int32(0)
+	}
+}
+
+func (a *autoscaler) Autoscale(ctx context.Context, attemptScaleDown bool, pending int32) {
+	a.trigger <- pending
+	a.syncAutoscale(ctx, attemptScaleDown, pending)
+}
+
+func (a *autoscaler) syncAutoscale(ctx context.Context, attemptScaleDown bool, pending int32) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+
+	wait.Poll(500*time.Millisecond, 5*time.Second, func() (bool, error) {
+		err := a.doautoscale(ctx, attemptScaleDown, pending)
+		return err == nil, nil
+	})
+}
+
+func (a *autoscaler) doautoscale(ctx context.Context, attemptScaleDown bool, pending int32) error {
+	state, err := a.stateAccessor.State(nil)
+	if err != nil {
+		a.logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
+		return err
+	}
+
+	scale, err := a.statefulSetClient.GetScale(ctx, a.statefulSetName, metav1.GetOptions{})
+	if err != nil {
+		// skip a beat
+		a.logger.Infow("failed to get scale subresource", zap.Error(err))
+		return err
+	}
+
+	a.logger.Infow("checking adapter capacity",
+		zap.Int32("pending", pending),
+		zap.Int32("replicas", scale.Spec.Replicas),
+		zap.Int32("last ordinal", state.LastOrdinal))
+
+	var scaleUpFactor, newreplicas, minNumPods int32
+	scaleUpFactor = 1                                                                                         // Non-HA scaling
+	if state.SchedPolicy != nil && contains(nil, state.SchedPolicy.Priorities, st.AvailabilityZonePriority) { //HA scaling across zones
+		scaleUpFactor = state.NumZones
+	}
+	if state.SchedPolicy != nil && contains(nil, state.SchedPolicy.Priorities, st.AvailabilityNodePriority) { //HA scaling across nodes
+		scaleUpFactor = state.NumNodes
+	}
+
+	newreplicas = state.LastOrdinal + 1 // Ideal number
+
+	// Take into account pending replicas and pods that are already filled (for even pod spread)
+	if pending > 0 {
+		// Make sure to allocate enough pods for holding all pending replicas.
+		if state.SchedPolicy != nil && contains(state.SchedPolicy.Predicates, nil, st.EvenPodSpread) && len(state.FreeCap) > 0 { //HA scaling across pods
+			leastNonZeroCapacity := a.minNonZeroInt(state.FreeCap)
+			minNumPods = int32(math.Ceil(float64(pending) / float64(leastNonZeroCapacity)))
+		} else {
+			minNumPods = int32(math.Ceil(float64(pending) / float64(a.capacity)))
+		}
+		newreplicas += int32(math.Ceil(float64(minNumPods)/float64(scaleUpFactor)) * float64(scaleUpFactor))
+	}
+
+	// Make sure to never scale down past the last ordinal
+	if newreplicas <= state.LastOrdinal {
+		newreplicas = state.LastOrdinal + scaleUpFactor
+	}
+
+	// Only scale down if permitted
+	if !attemptScaleDown && newreplicas < scale.Spec.Replicas {
+		newreplicas = scale.Spec.Replicas
+	}
+
+	if newreplicas != scale.Spec.Replicas {
+		scale.Spec.Replicas = newreplicas
+		a.logger.Infow("updating adapter replicas", zap.Int32("replicas", scale.Spec.Replicas))
+
+		_, err = a.statefulSetClient.UpdateScale(ctx, a.statefulSetName, scale, metav1.UpdateOptions{})
+		if err != nil {
+			a.logger.Errorw("updating scale subresource failed", zap.Error(err))
+			return err
+		}
+	} else if attemptScaleDown {
+		// since the number of replicas hasn't changed and time has approached to scale down,
+		// take the opportunity to compact the vreplicas
+		a.mayCompact(state, scaleUpFactor)
+	}
+	return nil
+}
+
+func (a *autoscaler) mayCompact(s *st.State, scaleUpFactor int32) {
+	// when there is only one pod there is nothing to move or number of pods is just enough!
+	if s.LastOrdinal < 1 || len(s.SchedulablePods) <= int(scaleUpFactor) {
+		return
+	}
+
+	if s.SchedulerPolicy == scheduler.MAXFILLUP {
+		// Determine if there is enough free capacity to
+		// move all vreplicas placed in the last pod to pods with a lower ordinal
+		freeCapacity := s.FreeCapacity() - s.Free(s.LastOrdinal)
+		usedInLastPod := s.Capacity - s.Free(s.LastOrdinal)
+
+		if freeCapacity >= usedInLastPod {
+			err := a.compact(s, scaleUpFactor)
+			if err != nil {
+				a.logger.Errorw("vreplicas compaction failed", zap.Error(err))
+			}
+		}
+
+		// only do 1 replica at a time to avoid overloading the scheduler with too many
+		// rescheduling requests.
+	} else if s.SchedPolicy != nil {
+		//Below calculation can be optimized to work for recovery scenarios when nodes/zones are lost due to failure
+		freeCapacity := s.FreeCapacity()
+		usedInLastXPods := s.Capacity * scaleUpFactor
+		for i := int32(0); i < scaleUpFactor && s.LastOrdinal-i >= 0; i++ {
+			freeCapacity = freeCapacity - s.Free(s.LastOrdinal-i)
+			usedInLastXPods = usedInLastXPods - s.Free(s.LastOrdinal-i)
+		}
+
+		if (freeCapacity >= usedInLastXPods) && //remaining pods can hold all vreps from evicted pods
+			(s.Replicas-scaleUpFactor >= scaleUpFactor) { //remaining # of pods is enough for HA scaling
+			err := a.compact(s, scaleUpFactor)
+			if err != nil {
+				a.logger.Errorw("vreplicas compaction failed", zap.Error(err))
+			}
+		}
+	}
+}
+
+func (a *autoscaler) compact(s *st.State, scaleUpFactor int32) error {
+	var pod *v1.Pod
+	vpods, err := a.vpodLister()
+	if err != nil {
+		return err
+	}
+
+	for _, vpod := range vpods {
+		placements := vpod.GetPlacements()
+		for i := len(placements) - 1; i >= 0; i-- { //start from the last placement
+			for j := int32(0); j < scaleUpFactor; j++ {
+				ordinal := st.OrdinalFromPodName(placements[i].PodName)
+
+				if ordinal == s.LastOrdinal-j {
+					wait.PollImmediate(50*time.Millisecond, 5*time.Second, func() (bool, error) {
+						if s.PodLister != nil {
+							pod, err = s.PodLister.Get(placements[i].PodName)
+						}
+						return err == nil, nil
+					})
+
+					err = a.evictor(pod, vpod, &placements[i])
+					if err != nil {
+						return err
+					}
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func contains(preds []scheduler.PredicatePolicy, priors []scheduler.PriorityPolicy, name string) bool {
+	for _, v := range preds {
+		if v.Name == name {
+			return true
+		}
+	}
+	for _, v := range priors {
+		if v.Name == name {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (a *autoscaler) minNonZeroInt(slice []int32) int32 {
+	min := a.capacity
+	for _, v := range slice {
+		if v < min && v > 0 {
+			min = v
+		}
+	}
+	return min
+}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/statefulset/scheduler.go b/vendor/knative.dev/eventing/pkg/scheduler/statefulset/scheduler.go
new file mode 100644
index 000000000..274547275
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/scheduler/statefulset/scheduler.go
@@ -0,0 +1,712 @@
+/*
+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"
+	"crypto/rand"
+	"fmt"
+	"math/big"
+	"sort"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+	appsv1 "k8s.io/api/apps/v1"
+	"k8s.io/apimachinery/pkg/types"
+	clientappsv1 "k8s.io/client-go/kubernetes/typed/apps/v1"
+	corev1listers "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/utils/integer"
+
+	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/pkg/apis/duck/v1alpha1"
+	"knative.dev/eventing/pkg/scheduler"
+	"knative.dev/eventing/pkg/scheduler/factory"
+	st "knative.dev/eventing/pkg/scheduler/state"
+	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
+
+	_ "knative.dev/eventing/pkg/scheduler/plugins/core/availabilitynodepriority"
+	_ "knative.dev/eventing/pkg/scheduler/plugins/core/availabilityzonepriority"
+	_ "knative.dev/eventing/pkg/scheduler/plugins/core/evenpodspread"
+	_ "knative.dev/eventing/pkg/scheduler/plugins/core/lowestordinalpriority"
+	_ "knative.dev/eventing/pkg/scheduler/plugins/core/podfitsresources"
+	_ "knative.dev/eventing/pkg/scheduler/plugins/core/removewithavailabilitynodepriority"
+	_ "knative.dev/eventing/pkg/scheduler/plugins/core/removewithavailabilityzonepriority"
+	_ "knative.dev/eventing/pkg/scheduler/plugins/core/removewithevenpodspreadpriority"
+	_ "knative.dev/eventing/pkg/scheduler/plugins/core/removewithhighestordinalpriority"
+	_ "knative.dev/eventing/pkg/scheduler/plugins/kafka/nomaxresourcecount"
+)
+
+// NewScheduler creates a new scheduler with pod autoscaling enabled.
+func NewScheduler(ctx context.Context,
+	namespace, name string,
+	lister scheduler.VPodLister,
+	refreshPeriod time.Duration,
+	capacity int32,
+	schedulerPolicy scheduler.SchedulerPolicyType,
+	nodeLister corev1listers.NodeLister,
+	evictor scheduler.Evictor,
+	schedPolicy *scheduler.SchedulerPolicy,
+	deschedPolicy *scheduler.SchedulerPolicy) scheduler.Scheduler {
+
+	podInformer := podinformer.Get(ctx)
+	podLister := podInformer.Lister().Pods(namespace)
+
+	stateAccessor := st.NewStateBuilder(ctx, namespace, name, lister, capacity, schedulerPolicy, schedPolicy, deschedPolicy, podLister, nodeLister)
+	autoscaler := NewAutoscaler(ctx, namespace, name, lister, stateAccessor, evictor, refreshPeriod, capacity)
+
+	go autoscaler.Start(ctx)
+
+	return NewStatefulSetScheduler(ctx, namespace, name, lister, stateAccessor, autoscaler, podLister)
+}
+
+// StatefulSetScheduler is a scheduler placing VPod into statefulset-managed set of pods
+type StatefulSetScheduler struct {
+	ctx               context.Context
+	logger            *zap.SugaredLogger
+	statefulSetName   string
+	statefulSetClient clientappsv1.StatefulSetInterface
+	podLister         corev1listers.PodNamespaceLister
+	vpodLister        scheduler.VPodLister
+	lock              sync.Locker
+	stateAccessor     st.StateAccessor
+	autoscaler        Autoscaler
+
+	// replicas is the (cached) number of statefulset replicas.
+	replicas int32
+
+	// pending tracks the number of virtual replicas that haven't been scheduled yet
+	// because there wasn't enough free capacity.
+	// The autoscaler uses
+	pending map[types.NamespacedName]int32
+
+	// reserved tracks vreplicas that have been placed (ie. scheduled) but haven't been
+	// committed yet (ie. not appearing in vpodLister)
+	reserved map[types.NamespacedName]map[string]int32
+}
+
+func NewStatefulSetScheduler(ctx context.Context,
+	namespace, name string,
+	lister scheduler.VPodLister,
+	stateAccessor st.StateAccessor,
+	autoscaler Autoscaler, podlister corev1listers.PodNamespaceLister) scheduler.Scheduler {
+
+	scheduler := &StatefulSetScheduler{
+		ctx:               ctx,
+		logger:            logging.FromContext(ctx),
+		statefulSetName:   name,
+		statefulSetClient: kubeclient.Get(ctx).AppsV1().StatefulSets(namespace),
+		podLister:         podlister,
+		vpodLister:        lister,
+		pending:           make(map[types.NamespacedName]int32),
+		lock:              new(sync.Mutex),
+		stateAccessor:     stateAccessor,
+		reserved:          make(map[types.NamespacedName]map[string]int32),
+		autoscaler:        autoscaler,
+	}
+
+	// Monitor our statefulset
+	statefulsetInformer := statefulsetinformer.Get(ctx)
+	statefulsetInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: controller.FilterWithNameAndNamespace(namespace, name),
+		Handler:    controller.HandleAll(scheduler.updateStatefulset),
+	})
+
+	return scheduler
+}
+
+func (s *StatefulSetScheduler) Schedule(vpod scheduler.VPod) ([]duckv1alpha1.Placement, error) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	vpods, err := s.vpodLister()
+	if err != nil {
+		return nil, err
+	}
+	vpodFromLister := st.GetVPod(vpod.GetKey(), vpods)
+	if vpod.GetResourceVersion() != vpodFromLister.GetResourceVersion() {
+		return nil, fmt.Errorf("vpod to schedule has resource version different from one in indexer")
+	}
+
+	placements, err := s.scheduleVPod(vpod)
+	if placements == nil {
+		return placements, err
+	}
+
+	sort.SliceStable(placements, func(i int, j int) bool {
+		return st.OrdinalFromPodName(placements[i].PodName) < st.OrdinalFromPodName(placements[j].PodName)
+	})
+
+	// Reserve new placements until they are committed to the vpod.
+	s.reservePlacements(vpod, placements)
+
+	return placements, err
+}
+
+func (s *StatefulSetScheduler) scheduleVPod(vpod scheduler.VPod) ([]duckv1alpha1.Placement, error) {
+	logger := s.logger.With("key", vpod.GetKey())
+	logger.Info("scheduling")
+
+	// Get the current placements state
+	// Quite an expensive operation but safe and simple.
+	state, err := s.stateAccessor.State(s.reserved)
+	if err != nil {
+		logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
+		return nil, err
+	}
+
+	placements := vpod.GetPlacements()
+	var left int32
+
+	// The scheduler when policy type is
+	// Policy: MAXFILLUP (SchedulerPolicyType == MAXFILLUP)
+	// - allocates as many vreplicas as possible to the same pod(s)
+	// - 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
+	}
+
+	if state.SchedulerPolicy != "" {
+		// 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)
+
+			// Do not trigger the autoscaler to avoid unnecessary churn
+
+			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(state, vpod.GetVReplicas()-tr, placements)
+
+	} else { //Predicates and priorities must be used for scheduling
+		// Need less => scale down
+		if tr > vpod.GetVReplicas() && state.DeschedPolicy != nil {
+			logger.Infow("scaling down", zap.Int32("vreplicas", tr), zap.Int32("new vreplicas", vpod.GetVReplicas()))
+			placements = s.removeReplicasWithPolicy(vpod, tr-vpod.GetVReplicas(), placements)
+
+			// Do not trigger the autoscaler to avoid unnecessary churn
+
+			return placements, nil
+		}
+
+		if state.SchedPolicy != nil {
+
+			// Need more => scale up
+			// rebalancing needed for all vreps most likely since there are pending vreps from previous reconciliation
+			// can fall here when vreps scaled up or after eviction
+			logger.Infow("scaling up with a rebalance (if needed)", zap.Int32("vreplicas", tr), zap.Int32("new vreplicas", vpod.GetVReplicas()))
+			placements, left = s.rebalanceReplicasWithPolicy(vpod, vpod.GetVReplicas(), 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), zap.Int32("left", left))
+
+		s.pending[vpod.GetKey()] = left
+
+		// Trigger the autoscaler
+		if s.autoscaler != nil {
+			s.autoscaler.Autoscale(s.ctx, false, s.pendingVReplicas())
+
+			/* 			if state.SchedPolicy != nil {
+				delete(s.pending, vpod.GetKey()) //rebalancing doesn't care about pending since all vreps will be re-placed
+			} */
+		}
+
+		return placements, scheduler.ErrNotEnoughReplicas
+	}
+
+	logger.Infow("scheduling successful", zap.Any("placement", placements))
+	delete(s.pending, vpod.GetKey())
+
+	return placements, nil
+}
+
+func (s *StatefulSetScheduler) rebalanceReplicasWithPolicy(vpod scheduler.VPod, diff int32, placements []duckv1alpha1.Placement) ([]duckv1alpha1.Placement, int32) {
+	newPlacements := make([]duckv1alpha1.Placement, len(placements))
+	for i := 0; i < len(placements); i++ {
+		newPlacements[i].PodName = placements[i].PodName
+		newPlacements[i].VReplicas = 0
+	}
+	s.reservePlacements(vpod, newPlacements)
+
+	placements, diff = s.addReplicasWithPolicy(vpod, diff, newPlacements)
+
+	return placements, diff
+}
+
+func (s *StatefulSetScheduler) removeReplicasWithPolicy(vpod scheduler.VPod, diff int32, placements []duckv1alpha1.Placement) []duckv1alpha1.Placement {
+	logger := s.logger.Named("remove replicas with policy")
+	numVreps := diff
+
+	for i := int32(0); i < numVreps; i++ { //deschedule one vreplica at a time
+		state, err := s.stateAccessor.State(s.reserved)
+		if err != nil {
+			logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
+			return placements
+		}
+
+		feasiblePods := s.findFeasiblePods(s.ctx, state, vpod, state.DeschedPolicy)
+		feasiblePods = s.removePodsNotInPlacement(vpod, feasiblePods)
+		if len(feasiblePods) == 1 { //nothing to score, remove vrep from that pod
+			placementPodID := feasiblePods[0]
+			logger.Infof("Selected pod #%v to remove vreplica #%v from", placementPodID, i)
+			placements = s.removeSelectionFromPlacements(placementPodID, placements)
+			//state.SetFree(placementPodID, state.Free(placementPodID)+1)
+			s.reservePlacements(vpod, placements)
+			continue
+		}
+
+		priorityList, err := s.prioritizePods(s.ctx, state, vpod, feasiblePods, state.DeschedPolicy)
+		if err != nil {
+			logger.Info("error while scoring pods using priorities", zap.Error(err))
+			s.reservePlacements(vpod, placements)
+			break
+		}
+
+		placementPodID, err := s.selectPod(priorityList)
+		if err != nil {
+			logger.Info("error while selecting the placement pod", zap.Error(err))
+			s.reservePlacements(vpod, placements)
+			break
+		}
+
+		logger.Infof("Selected pod #%v to remove vreplica #%v from", placementPodID, i)
+		placements = s.removeSelectionFromPlacements(placementPodID, placements)
+		//state.SetFree(placementPodID, state.Free(placementPodID)+1)
+		s.reservePlacements(vpod, placements)
+	}
+	return placements
+}
+
+func (s *StatefulSetScheduler) removeSelectionFromPlacements(placementPodID int32, placements []duckv1alpha1.Placement) []duckv1alpha1.Placement {
+	newPlacements := make([]duckv1alpha1.Placement, 0, len(placements))
+
+	for i := 0; i < len(placements); i++ {
+		ordinal := st.OrdinalFromPodName(placements[i].PodName)
+		if placementPodID == ordinal {
+			if placements[i].VReplicas == 1 {
+				// remove the entire placement
+			} else {
+				newPlacements = append(newPlacements, duckv1alpha1.Placement{
+					PodName:   placements[i].PodName,
+					VReplicas: placements[i].VReplicas - 1,
+				})
+			}
+		} else {
+			newPlacements = append(newPlacements, duckv1alpha1.Placement{
+				PodName:   placements[i].PodName,
+				VReplicas: placements[i].VReplicas,
+			})
+		}
+	}
+	return newPlacements
+}
+
+func (s *StatefulSetScheduler) addReplicasWithPolicy(vpod scheduler.VPod, diff int32, placements []duckv1alpha1.Placement) ([]duckv1alpha1.Placement, int32) {
+	logger := s.logger.Named("add replicas with policy")
+
+	numVreps := diff
+	for i := int32(0); i < numVreps; i++ { //schedule one vreplica at a time (find most suitable pod placement satisying predicates with high score)
+		// Get the current placements state
+		state, err := s.stateAccessor.State(s.reserved)
+		if err != nil {
+			logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
+			return placements, diff
+		}
+
+		if s.replicas == 0 { //no pods to filter
+			logger.Infow("no pods available in statefulset")
+			s.reservePlacements(vpod, placements)
+			diff = numVreps - i //for autoscaling up
+			break               //end the iteration for all vreps since there are not pods
+		}
+
+		feasiblePods := s.findFeasiblePods(s.ctx, state, vpod, state.SchedPolicy)
+		if len(feasiblePods) == 0 { //no pods available to schedule this vreplica
+			logger.Info("no feasible pods available to schedule this vreplica")
+			s.reservePlacements(vpod, placements)
+			diff = numVreps - i //for autoscaling up and possible rebalancing
+			break
+		}
+
+		/* 	if len(feasiblePods) == 1 { //nothing to score, place vrep on that pod (Update: for HA, must run HA scorers)
+			placementPodID := feasiblePods[0]
+			logger.Infof("Selected pod #%v for vreplica #%v ", placementPodID, i)
+			placements = s.addSelectionToPlacements(placementPodID, placements)
+			//state.SetFree(placementPodID, state.Free(placementPodID)-1)
+			s.reservePlacements(vpod, placements)
+			diff--
+			continue
+		} */
+
+		priorityList, err := s.prioritizePods(s.ctx, state, vpod, feasiblePods, state.SchedPolicy)
+		if err != nil {
+			logger.Info("error while scoring pods using priorities", zap.Error(err))
+			s.reservePlacements(vpod, placements)
+			diff = numVreps - i //for autoscaling up and possible rebalancing
+			break
+		}
+
+		placementPodID, err := s.selectPod(priorityList)
+		if err != nil {
+			logger.Info("error while selecting the placement pod", zap.Error(err))
+			s.reservePlacements(vpod, placements)
+			diff = numVreps - i //for autoscaling up and possible rebalancing
+			break
+		}
+
+		logger.Infof("Selected pod #%v for vreplica #%v", placementPodID, i)
+		placements = s.addSelectionToPlacements(placementPodID, placements)
+		//state.SetFree(placementPodID, state.Free(placementPodID)-1)
+		s.reservePlacements(vpod, placements)
+		diff--
+	}
+	return placements, diff
+}
+
+func (s *StatefulSetScheduler) addSelectionToPlacements(placementPodID int32, placements []duckv1alpha1.Placement) []duckv1alpha1.Placement {
+	seen := false
+
+	for i := 0; i < len(placements); i++ {
+		ordinal := st.OrdinalFromPodName(placements[i].PodName)
+		if placementPodID == ordinal {
+			seen = true
+			placements[i].VReplicas = placements[i].VReplicas + 1
+		}
+	}
+	if !seen {
+		placements = append(placements, duckv1alpha1.Placement{
+			PodName:   st.PodNameFromOrdinal(s.statefulSetName, placementPodID),
+			VReplicas: 1,
+		})
+	}
+	return placements
+}
+
+// findFeasiblePods finds the pods that fit the filter plugins
+func (s *StatefulSetScheduler) findFeasiblePods(ctx context.Context, state *st.State, vpod scheduler.VPod, policy *scheduler.SchedulerPolicy) []int32 {
+	feasiblePods := make([]int32, 0)
+	for _, podId := range state.SchedulablePods {
+		statusMap := s.RunFilterPlugins(ctx, state, vpod, podId, policy)
+		status := statusMap.Merge()
+		if status.IsSuccess() {
+			feasiblePods = append(feasiblePods, podId)
+		}
+	}
+
+	return feasiblePods
+}
+
+// removePodsNotInPlacement removes pods that do not have vreplicas placed
+func (s *StatefulSetScheduler) removePodsNotInPlacement(vpod scheduler.VPod, feasiblePods []int32) []int32 {
+	newFeasiblePods := make([]int32, 0)
+	for _, e := range vpod.GetPlacements() {
+		for _, podID := range feasiblePods {
+			if podID == st.OrdinalFromPodName(e.PodName) { //if pod is in current placement list
+				newFeasiblePods = append(newFeasiblePods, podID)
+			}
+		}
+	}
+
+	return newFeasiblePods
+}
+
+// prioritizePods prioritizes the pods by running the score plugins, which return a score for each pod.
+// The scores from each plugin are added together to make the score for that pod.
+func (s *StatefulSetScheduler) prioritizePods(ctx context.Context, states *st.State, vpod scheduler.VPod, feasiblePods []int32, policy *scheduler.SchedulerPolicy) (st.PodScoreList, error) {
+	logger := s.logger.Named("prioritize all feasible pods")
+
+	// If no priority configs are provided, then all pods will have a score of one
+	result := make(st.PodScoreList, 0, len(feasiblePods))
+	if !s.HasScorePlugins(states, policy) {
+		for _, podID := range feasiblePods {
+			result = append(result, st.PodScore{
+				ID:    podID,
+				Score: 1,
+			})
+		}
+		return result, nil
+	}
+
+	scoresMap, scoreStatus := s.RunScorePlugins(ctx, states, vpod, feasiblePods, policy)
+	if !scoreStatus.IsSuccess() {
+		logger.Infof("FAILURE! Cannot score feasible pods due to plugin errors %v", scoreStatus.AsError())
+		return nil, scoreStatus.AsError()
+	}
+
+	// Summarize all scores.
+	for i := range feasiblePods {
+		result = append(result, st.PodScore{ID: feasiblePods[i], Score: 0})
+		for j := range scoresMap {
+			result[i].Score += scoresMap[j][i].Score
+		}
+	}
+
+	return result, nil
+}
+
+// selectPod takes a prioritized list of pods and then picks one
+func (s *StatefulSetScheduler) selectPod(podScoreList st.PodScoreList) (int32, error) {
+	if len(podScoreList) == 0 {
+		return -1, fmt.Errorf("empty priority list") //no selected pod
+	}
+
+	maxScore := podScoreList[0].Score
+	selected := podScoreList[0].ID
+	cntOfMaxScore := int64(1)
+	for _, ps := range podScoreList[1:] {
+		if ps.Score > maxScore {
+			maxScore = ps.Score
+			selected = ps.ID
+			cntOfMaxScore = 1
+		} else if ps.Score == maxScore { //if equal scores, randomly picks one
+			cntOfMaxScore++
+			randNum, err := rand.Int(rand.Reader, big.NewInt(cntOfMaxScore))
+			if err != nil {
+				return -1, fmt.Errorf("failed to generate random number")
+			}
+			if randNum.Int64() == int64(0) {
+				selected = ps.ID
+			}
+		}
+	}
+	return selected, nil
+}
+
+// RunFilterPlugins runs the set of configured Filter plugins for a vrep on the given pod.
+// If any of these plugins doesn't return "Success", the pod is not suitable for placing the vrep.
+// Meanwhile, the failure message and status are set for the given pod.
+func (s *StatefulSetScheduler) RunFilterPlugins(ctx context.Context, states *st.State, vpod scheduler.VPod, podID int32, policy *scheduler.SchedulerPolicy) st.PluginToStatus {
+	logger := s.logger.Named("run all filter plugins")
+
+	statuses := make(st.PluginToStatus)
+	for _, plugin := range policy.Predicates {
+		pl, err := factory.GetFilterPlugin(plugin.Name)
+		if err != nil {
+			logger.Error("Could not find filter plugin in Registry: ", plugin.Name)
+			continue
+		}
+
+		//logger.Infof("Going to run filter plugin: %s using state: %v ", pl.Name(), states)
+		pluginStatus := s.runFilterPlugin(ctx, pl, plugin.Args, states, vpod, podID)
+		if !pluginStatus.IsSuccess() {
+			if !pluginStatus.IsUnschedulable() {
+				errStatus := st.NewStatus(st.Error, fmt.Sprintf("running %q filter plugin for pod %q failed with: %v", pl.Name(), podID, pluginStatus.Message()))
+				return map[string]*st.Status{pl.Name(): errStatus} //TODO: if one plugin fails, then no more plugins are run
+			}
+			statuses[pl.Name()] = pluginStatus
+			return statuses
+		}
+	}
+
+	return statuses
+}
+
+func (s *StatefulSetScheduler) runFilterPlugin(ctx context.Context, pl st.FilterPlugin, args interface{}, states *st.State, vpod scheduler.VPod, podID int32) *st.Status {
+	status := pl.Filter(ctx, args, states, vpod.GetKey(), podID)
+	return status
+}
+
+// RunScorePlugins runs the set of configured scoring plugins. It returns a list that stores for each scoring plugin name the corresponding PodScoreList(s).
+// It also returns *Status, which is set to non-success if any of the plugins returns a non-success status.
+func (s *StatefulSetScheduler) RunScorePlugins(ctx context.Context, states *st.State, vpod scheduler.VPod, feasiblePods []int32, policy *scheduler.SchedulerPolicy) (st.PluginToPodScores, *st.Status) {
+	logger := s.logger.Named("run all score plugins")
+
+	pluginToPodScores := make(st.PluginToPodScores, len(policy.Priorities))
+	for _, plugin := range policy.Priorities {
+		pl, err := factory.GetScorePlugin(plugin.Name)
+		if err != nil {
+			logger.Error("Could not find score plugin in registry: ", plugin.Name)
+			continue
+		}
+
+		//logger.Infof("Going to run score plugin: %s using state: %v ", pl.Name(), states)
+		pluginToPodScores[pl.Name()] = make(st.PodScoreList, len(feasiblePods))
+		for index, podID := range feasiblePods {
+			score, pluginStatus := s.runScorePlugin(ctx, pl, plugin.Args, states, feasiblePods, vpod, podID)
+			if !pluginStatus.IsSuccess() {
+				errStatus := st.NewStatus(st.Error, fmt.Sprintf("running %q scoring plugin for pod %q failed with: %v", pl.Name(), podID, pluginStatus.AsError()))
+				return pluginToPodScores, errStatus //TODO: if one plugin fails, then no more plugins are run
+			}
+
+			score = score * plugin.Weight //WEIGHED SCORE VALUE
+			//logger.Infof("scoring plugin %q produced score %v for pod %q: %v", pl.Name(), score, podID, pluginStatus)
+			pluginToPodScores[pl.Name()][index] = st.PodScore{
+				ID:    podID,
+				Score: score,
+			}
+		}
+
+		status := pl.ScoreExtensions().NormalizeScore(ctx, states, pluginToPodScores[pl.Name()]) //NORMALIZE SCORES FOR ALL FEASIBLE PODS
+		if !status.IsSuccess() {
+			errStatus := st.NewStatus(st.Error, fmt.Sprintf("running %q scoring plugin failed with: %v", pl.Name(), status.AsError()))
+			return pluginToPodScores, errStatus
+		}
+	}
+
+	return pluginToPodScores, st.NewStatus(st.Success)
+}
+
+func (s *StatefulSetScheduler) runScorePlugin(ctx context.Context, pl st.ScorePlugin, args interface{}, states *st.State, feasiblePods []int32, vpod scheduler.VPod, podID int32) (uint64, *st.Status) {
+	score, status := pl.Score(ctx, args, states, feasiblePods, vpod.GetKey(), podID)
+	return score, status
+}
+
+// HasScorePlugins returns true if at least one score plugin is defined.
+func (s *StatefulSetScheduler) HasScorePlugins(state *st.State, policy *scheduler.SchedulerPolicy) bool {
+	return len(policy.Priorities) > 0
+}
+
+func (s *StatefulSetScheduler) removeReplicas(diff int32, placements []duckv1alpha1.Placement) []duckv1alpha1.Placement {
+	newPlacements := make([]duckv1alpha1.Placement, 0, len(placements))
+	for i := len(placements) - 1; i > -1; 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(states *st.State, diff int32, placements []duckv1alpha1.Placement) ([]duckv1alpha1.Placement, int32) {
+	// Pod affinity algorithm: prefer adding replicas to existing pods before considering other replicas
+	newPlacements := make([]duckv1alpha1.Placement, 0, len(placements))
+
+	// Add to existing
+	for i := 0; i < len(placements); i++ {
+		podName := placements[i].PodName
+		ordinal := st.OrdinalFromPodName(podName)
+
+		// Is there space in PodName?
+		f := states.Free(ordinal)
+		if diff >= 0 && f > 0 {
+			allocation := integer.Int32Min(f, diff)
+			newPlacements = append(newPlacements, duckv1alpha1.Placement{
+				PodName:   podName,
+				VReplicas: placements[i].VReplicas + allocation,
+			})
+
+			diff -= allocation
+			states.SetFree(ordinal, f-allocation)
+		} else {
+			newPlacements = append(newPlacements, placements[i])
+		}
+	}
+
+	if diff > 0 {
+		// Needs to allocate replicas to additional pods
+		for ordinal := int32(0); ordinal < s.replicas; ordinal++ {
+			f := states.Free(ordinal)
+			if f > 0 {
+				allocation := integer.Int32Min(f, diff)
+				newPlacements = append(newPlacements, duckv1alpha1.Placement{
+					PodName:   st.PodNameFromOrdinal(s.statefulSetName, ordinal),
+					VReplicas: allocation,
+				})
+
+				diff -= allocation
+				states.SetFree(ordinal, f-allocation)
+			}
+
+			if diff == 0 {
+				break
+			}
+		}
+	}
+
+	return newPlacements, diff
+}
+
+// pendingReplicas returns the total number of vreplicas
+// that haven't been scheduled yet
+func (s *StatefulSetScheduler) pendingVReplicas() int32 {
+	t := int32(0)
+	for _, v := range s.pending {
+		t += v
+	}
+	return t
+}
+
+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 statefulset.Spec.Replicas == nil {
+		s.replicas = 1
+	} else if s.replicas != *statefulset.Spec.Replicas {
+		s.replicas = *statefulset.Spec.Replicas
+		s.logger.Infow("statefulset replicas updated", zap.Int32("replicas", s.replicas))
+	}
+}
+
+func (s *StatefulSetScheduler) reservePlacements(vpod scheduler.VPod, placements []duckv1alpha1.Placement) {
+	existing := vpod.GetPlacements()
+	for _, p := range placements {
+		placed := int32(0)
+		for _, e := range existing {
+			if e.PodName == p.PodName {
+				placed = e.VReplicas
+				break
+			}
+		}
+
+		// Only record placements exceeding existing ones, since the
+		// goal is to prevent pods to be overcommitted.
+		// When less than existing ones, recording is done to help with
+		// for removal of vreps with descheduling policies
+		// note: track all vreplicas, not only the new ones since
+		// the next time `state()` is called some vreplicas might
+		// have been committed.
+		if placed != p.VReplicas {
+			if _, ok := s.reserved[vpod.GetKey()]; !ok {
+				s.reserved[vpod.GetKey()] = make(map[string]int32)
+			}
+			s.reserved[vpod.GetKey()][p.PodName] = p.VReplicas
+		}
+	}
+
+}
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..518d654ec
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake/fake.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// 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..76806dec7
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/statefulset.go
@@ -0,0 +1,116 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package statefulset
+
+import (
+	context "context"
+
+	apiappsv1 "k8s.io/api/apps/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	v1 "k8s.io/client-go/informers/apps/v1"
+	kubernetes "k8s.io/client-go/kubernetes"
+	appsv1 "k8s.io/client-go/listers/apps/v1"
+	cache "k8s.io/client-go/tools/cache"
+	client "knative.dev/pkg/client/injection/kube/client"
+	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)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// 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()
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	inf := &wrapper{client: client.Get(ctx), resourceVersion: injection.GetResourceVersion(ctx)}
+	return context.WithValue(ctx, Key{}, inf)
+}
+
+// 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)
+}
+
+type wrapper struct {
+	client kubernetes.Interface
+
+	namespace string
+
+	resourceVersion string
+}
+
+var _ v1.StatefulSetInformer = (*wrapper)(nil)
+var _ appsv1.StatefulSetLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apiappsv1.StatefulSet{}, 0, nil)
+}
+
+func (w *wrapper) Lister() appsv1.StatefulSetLister {
+	return w
+}
+
+func (w *wrapper) StatefulSets(namespace string) appsv1.StatefulSetNamespaceLister {
+	return &wrapper{client: w.client, namespace: namespace, resourceVersion: w.resourceVersion}
+}
+
+// SetResourceVersion allows consumers to adjust the minimum resourceVersion
+// used by the underlying client.  It is not accessible via the standard
+// lister interface, but can be accessed through a user-defined interface and
+// an implementation check e.g. rvs, ok := foo.(ResourceVersionSetter)
+func (w *wrapper) SetResourceVersion(resourceVersion string) {
+	w.resourceVersion = resourceVersion
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apiappsv1.StatefulSet, err error) {
+	lo, err := w.client.AppsV1().StatefulSets(w.namespace).List(context.TODO(), metav1.ListOptions{
+		LabelSelector:   selector.String(),
+		ResourceVersion: w.resourceVersion,
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apiappsv1.StatefulSet, error) {
+	return w.client.AppsV1().StatefulSets(w.namespace).Get(context.TODO(), name, metav1.GetOptions{
+		ResourceVersion: w.resourceVersion,
+	})
+}
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/statefulset_expansion.go b/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/statefulset_expansion.go
new file mode 100644
index 000000000..699b311d0
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/statefulset_expansion.go
@@ -0,0 +1,26 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package statefulset
+
+import (
+	apps "k8s.io/api/apps/v1"
+	v1 "k8s.io/api/core/v1"
+)
+
+func (w *wrapper) GetPodStatefulSets(pod *v1.Pod) ([]*apps.StatefulSet, error) {
+	panic("NYI")
+}
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake/fake.go
new file mode 100644
index 000000000..627d83d7c
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake/fake.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	node "knative.dev/pkg/client/injection/kube/informers/core/v1/node"
+	fake "knative.dev/pkg/client/injection/kube/informers/factory/fake"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = node.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Core().V1().Nodes()
+	return context.WithValue(ctx, node.Key{}, inf), inf.Informer()
+}
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/node.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/node.go
new file mode 100644
index 000000000..b4712a000
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/node/node.go
@@ -0,0 +1,110 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package node
+
+import (
+	context "context"
+
+	apicorev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	v1 "k8s.io/client-go/informers/core/v1"
+	kubernetes "k8s.io/client-go/kubernetes"
+	corev1 "k8s.io/client-go/listers/core/v1"
+	cache "k8s.io/client-go/tools/cache"
+	client "knative.dev/pkg/client/injection/kube/client"
+	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)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// 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.Core().V1().Nodes()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	inf := &wrapper{client: client.Get(ctx), resourceVersion: injection.GetResourceVersion(ctx)}
+	return context.WithValue(ctx, Key{}, inf)
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1.NodeInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch k8s.io/client-go/informers/core/v1.NodeInformer from context.")
+	}
+	return untyped.(v1.NodeInformer)
+}
+
+type wrapper struct {
+	client kubernetes.Interface
+
+	resourceVersion string
+}
+
+var _ v1.NodeInformer = (*wrapper)(nil)
+var _ corev1.NodeLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apicorev1.Node{}, 0, nil)
+}
+
+func (w *wrapper) Lister() corev1.NodeLister {
+	return w
+}
+
+// SetResourceVersion allows consumers to adjust the minimum resourceVersion
+// used by the underlying client.  It is not accessible via the standard
+// lister interface, but can be accessed through a user-defined interface and
+// an implementation check e.g. rvs, ok := foo.(ResourceVersionSetter)
+func (w *wrapper) SetResourceVersion(resourceVersion string) {
+	w.resourceVersion = resourceVersion
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apicorev1.Node, err error) {
+	lo, err := w.client.CoreV1().Nodes().List(context.TODO(), metav1.ListOptions{
+		LabelSelector:   selector.String(),
+		ResourceVersion: w.resourceVersion,
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apicorev1.Node, error) {
+	return w.client.CoreV1().Nodes().Get(context.TODO(), name, metav1.GetOptions{
+		ResourceVersion: w.resourceVersion,
+	})
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 1b0c9f168..5ba7d955c 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1196,6 +1196,19 @@ knative.dev/eventing/pkg/reconciler/testing
 knative.dev/eventing/pkg/reconciler/testing/scheme
 knative.dev/eventing/pkg/reconciler/testing/v1
 knative.dev/eventing/pkg/scheduler
+knative.dev/eventing/pkg/scheduler/factory
+knative.dev/eventing/pkg/scheduler/plugins/core/availabilitynodepriority
+knative.dev/eventing/pkg/scheduler/plugins/core/availabilityzonepriority
+knative.dev/eventing/pkg/scheduler/plugins/core/evenpodspread
+knative.dev/eventing/pkg/scheduler/plugins/core/lowestordinalpriority
+knative.dev/eventing/pkg/scheduler/plugins/core/podfitsresources
+knative.dev/eventing/pkg/scheduler/plugins/core/removewithavailabilitynodepriority
+knative.dev/eventing/pkg/scheduler/plugins/core/removewithavailabilityzonepriority
+knative.dev/eventing/pkg/scheduler/plugins/core/removewithevenpodspreadpriority
+knative.dev/eventing/pkg/scheduler/plugins/core/removewithhighestordinalpriority
+knative.dev/eventing/pkg/scheduler/plugins/kafka/nomaxresourcecount
+knative.dev/eventing/pkg/scheduler/state
+knative.dev/eventing/pkg/scheduler/statefulset
 knative.dev/eventing/pkg/utils
 knative.dev/eventing/test
 knative.dev/eventing/test/conformance/helpers
@@ -1344,8 +1357,12 @@ knative.dev/pkg/client/injection/kube/client
 knative.dev/pkg/client/injection/kube/client/fake
 knative.dev/pkg/client/injection/kube/informers/admissionregistration/v1/mutatingwebhookconfiguration
 knative.dev/pkg/client/injection/kube/informers/admissionregistration/v1/validatingwebhookconfiguration
+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/configmap
 knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake
+knative.dev/pkg/client/injection/kube/informers/core/v1/node
+knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/pod
 knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/secret


Test patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types_test.go
new file mode 100644
index 000000000..f44f42c35
--- /dev/null
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types_test.go
@@ -0,0 +1,126 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1alpha1
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestConsumerGroup_GetUserFacingResourceRef(t *testing.T) {
+	tests := []struct {
+		name       string
+		ObjectMeta metav1.ObjectMeta
+		want       *metav1.OwnerReference
+	}{
+		{
+			name:       "no owner ref",
+			ObjectMeta: metav1.ObjectMeta{},
+			want:       nil,
+		},
+		{
+			name: "no known owner ref",
+			ObjectMeta: metav1.ObjectMeta{
+				OwnerReferences: []metav1.OwnerReference{
+					{
+						APIVersion: "v1",
+						Kind:       "Pod",
+						Name:       "name",
+					},
+				},
+			},
+			want: nil,
+		},
+		{
+			name: "trigger owner ref",
+			ObjectMeta: metav1.ObjectMeta{
+				OwnerReferences: []metav1.OwnerReference{
+					{
+						APIVersion: "v1",
+						Kind:       "Pod",
+						Name:       "name",
+					},
+					{
+						APIVersion: "eventing.knative.dev/v1",
+						Kind:       "Trigger",
+						Name:       "name",
+					},
+				},
+			},
+			want: &metav1.OwnerReference{
+				APIVersion: "eventing.knative.dev/v1",
+				Kind:       "Trigger",
+				Name:       "name",
+			},
+		},
+		{
+			name: "kafkasource owner ref",
+			ObjectMeta: metav1.ObjectMeta{
+				OwnerReferences: []metav1.OwnerReference{
+					{
+						APIVersion: "v1",
+						Kind:       "Pod",
+						Name:       "name",
+					},
+					{
+						APIVersion: "sources.knative.dev/v1",
+						Kind:       "KafkaSource",
+						Name:       "name",
+					},
+				},
+			},
+			want: &metav1.OwnerReference{
+				APIVersion: "sources.knative.dev/v1",
+				Kind:       "KafkaSource",
+				Name:       "name",
+			},
+		},
+		{
+			name: "kafkachannel owner ref",
+			ObjectMeta: metav1.ObjectMeta{
+				OwnerReferences: []metav1.OwnerReference{
+					{
+						APIVersion: "v1",
+						Kind:       "Pod",
+						Name:       "name",
+					},
+					{
+						APIVersion: "messaging.knative.dev/v1beta1",
+						Kind:       "KafkaChannel",
+						Name:       "name",
+					},
+				},
+			},
+			want: &metav1.OwnerReference{
+				APIVersion: "messaging.knative.dev/v1beta1",
+				Kind:       "KafkaChannel",
+				Name:       "name",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cg := &ConsumerGroup{ObjectMeta: tt.ObjectMeta}
+			got := cg.GetUserFacingResourceRef()
+			if diff := cmp.Diff(tt.want, got); diff != "" {
+				t.Error("(-want, +got)", diff)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types_test.go
new file mode 100644
index 000000000..4ff7f61a9
--- /dev/null
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types_test.go
@@ -0,0 +1,82 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1alpha1
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestConsumer_GetConsumerGroup(t *testing.T) {
+	tests := []struct {
+		name       string
+		ObjectMeta metav1.ObjectMeta
+		want       *metav1.OwnerReference
+	}{
+		{
+			name:       "no owner reference",
+			ObjectMeta: metav1.ObjectMeta{},
+			want:       nil,
+		},
+		{
+			name: "no CG owner reference",
+			ObjectMeta: metav1.ObjectMeta{
+				OwnerReferences: []metav1.OwnerReference{
+					{
+						APIVersion: "v1",
+						Kind:       "Pod",
+						Name:       "name",
+					},
+				},
+			},
+			want: nil,
+		},
+		{
+			name: "CG owner reference",
+			ObjectMeta: metav1.ObjectMeta{
+				OwnerReferences: []metav1.OwnerReference{
+					{
+						APIVersion: "v1",
+						Kind:       "Pod",
+						Name:       "name",
+					},
+					{
+						APIVersion: SchemeGroupVersion.String(),
+						Kind:       ConsumerGroupGroupVersionKind.Kind,
+						Name:       "cg-name",
+					},
+				},
+			},
+			want: &metav1.OwnerReference{
+				APIVersion: SchemeGroupVersion.String(),
+				Kind:       ConsumerGroupGroupVersionKind.Kind,
+				Name:       "cg-name",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := &Consumer{ObjectMeta: tt.ObjectMeta}
+			got := c.GetConsumerGroup()
+			if diff := cmp.Diff(tt.want, got); diff != "" {
+				t.Error("(-want, +got)", diff)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index 15f57d27e..81f9e3ec3 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -67,6 +67,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 					)),
 					ConsumerGroupReplicas(2),
+					ConsumerForTrigger(),
 				),
 			},
 			Key: ConsumerGroupTestKey,
@@ -114,6 +115,7 @@ func TestReconcileKind(t *testing.T) {
 								),
 							)),
 							ConsumerGroupReplicas(2),
+							ConsumerForTrigger(),
 						)
 						cg.Status.Placements = []eventingduckv1alpha1.Placement{
 							{PodName: "p1", VReplicas: 1},
@@ -151,6 +153,7 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerGroupIdConfig("my.group.id"),
 						),
 					)),
+					ConsumerForTrigger(),
 					ConsumerGroupReplicas(2),
 				),
 			},
@@ -188,6 +191,7 @@ func TestReconcileKind(t *testing.T) {
 								),
 							)),
 							ConsumerGroupReplicas(2),
+							ConsumerForTrigger(),
 						)
 						cg.Status.Placements = []eventingduckv1alpha1.Placement{
 							{PodName: "p1", VReplicas: 1},
@@ -226,6 +230,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 					)),
 					ConsumerGroupReplicas(3),
+					ConsumerForTrigger(),
 				),
 			},
 			Key: ConsumerGroupTestKey,
@@ -283,6 +288,7 @@ func TestReconcileKind(t *testing.T) {
 								),
 							)),
 							ConsumerGroupReplicas(3),
+							ConsumerForTrigger(),
 						)
 						cg.Status.Placements = []eventingduckv1alpha1.Placement{
 							{PodName: "p1", VReplicas: 1},
@@ -319,6 +325,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 					)),
 					ConsumerGroupReplicas(2),
+					ConsumerForTrigger(),
 				),
 			},
 			Key: ConsumerGroupTestKey,
@@ -379,6 +386,7 @@ func TestReconcileKind(t *testing.T) {
 								),
 							)),
 							ConsumerGroupReplicas(2),
+							ConsumerForTrigger(),
 						)
 						cg.Status.Placements = []eventingduckv1alpha1.Placement{
 							{PodName: "p1", VReplicas: 1},
@@ -392,7 +400,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 		},
 		{
-			Name: "Scheduler failed",
+			Name: "Schedulers failed",
 			Objects: []runtime.Object{
 				NewConsumerGroup(
 					ConsumerGroupConsumerSpec(NewConsumerSpec(
@@ -403,6 +411,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 					)),
 					ConsumerGroupReplicas(2),
+					ConsumerForTrigger(),
 				),
 			},
 			Key: ConsumerGroupTestKey,
@@ -427,6 +436,7 @@ func TestReconcileKind(t *testing.T) {
 								),
 							)),
 							ConsumerGroupReplicas(2),
+							ConsumerForTrigger(),
 						)
 						cg.GetConditionSet().Manage(cg.GetStatus()).InitializeConditions()
 						_ = cg.MarkScheduleConsumerFailed("Schedule", io.EOF)
@@ -440,7 +450,9 @@ func TestReconcileKind(t *testing.T) {
 	tt.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
 		r := Reconciler{
-			Scheduler:       row.OtherTestData[testSchedulerKey].(scheduler.Scheduler),
+			SchedulerFunc: func(s string) scheduler.Scheduler {
+				return row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+			},
 			ConsumerLister:  listers.GetConsumerLister(),
 			InternalsClient: fakekafkainternalsclient.Get(ctx).InternalV1alpha1(),
 			NameGenerator:   &CounterGenerator{},
diff --git a/control-plane/pkg/reconciler/consumergroup/controller_test.go b/control-plane/pkg/reconciler/consumergroup/controller_test.go
new file mode 100644
index 000000000..3ee7d7a21
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumergroup/controller_test.go
@@ -0,0 +1,114 @@
+/*
+ * Copyright 2022 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 consumergroup
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake"
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumer/fake"
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
+)
+
+func TestNewController(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	t.Setenv("SYSTEM_NAMESPACE", systemNamespace)
+
+	ctx, _ = kubeclient.With(ctx,
+		&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      ConfigKafkaSchedulerName,
+				Namespace: systemNamespace,
+			},
+			Data: map[string]string{
+				"predicates": `
+			     [
+			       {"Name": "PodFitsResources"},
+			       {"Name": "NoMaxResourceCount", "Args": "{\"NumPartitions\": 100}"},
+			       {"Name": "EvenPodSpread", "Args": "{\"MaxSkew\": 2}"}
+			     ]`,
+				"priorities": `
+                [
+                  {"Name": "AvailabilityZonePriority", "Weight": 10, "Args":  "{\"MaxSkew\": 2}"},
+                  {"Name": "LowestOrdinalPriority", "Weight": 2}
+                ]`,
+			},
+		},
+		&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      ConfigKafkaDeSchedulerName,
+				Namespace: systemNamespace,
+			},
+			Data: map[string]string{
+				"predicates": `[]`,
+				"priorities": `
+                 [
+                    {"Name": "RemoveWithEvenPodSpreadPriority", "Weight": 10, "Args": "{\"MaxSkew\": 2}"},
+                    {"Name": "RemoveWithAvailabilityZonePriority", "Weight": 10, "Args":  "{\"MaxSkew\": 2}"},
+                    {"Name": "RemoveWithHighestOrdinalPriority", "Weight": 2}
+                 ]`,
+			},
+		},
+	)
+
+	controller := NewController(ctx)
+	if controller == nil {
+		t.Error("failed to create controller: <nil>")
+	}
+}
+
+func TestEnqueueConsumerFromConsumerGroup(t *testing.T) {
+
+	capture := types.NamespacedName{}
+
+	f := enqueueConsumerGroupFromConsumer(func(name types.NamespacedName) {
+		// Make sure this is called only once.
+		require.Empty(t, capture.Namespace)
+		require.Empty(t, capture.Name)
+
+		capture = name
+	})
+
+	ns := "ns"
+	or := metav1.OwnerReference{
+		APIVersion: kafkainternals.SchemeGroupVersion.String(),
+		Kind:       kafkainternals.ConsumerGroupGroupVersionKind.Kind,
+		Name:       "hello",
+	}
+
+	f(&kafkainternals.Consumer{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:       ns,
+			OwnerReferences: []metav1.OwnerReference{or},
+		},
+	})
+
+	require.Equal(t, or.Name, capture.Name)
+	require.Equal(t, ns, capture.Namespace)
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_consumer.go b/control-plane/pkg/reconciler/testing/objects_consumer.go
index e7db0e0eb..4797e71f5 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -181,3 +181,14 @@ func ConsumerFilters(filters *kafkainternals.Filters) ConsumerSpecOption {
 		c.Filters = filters
 	}
 }
+
+func ConsumerForTrigger() ConsumerGroupOption {
+	return func(c *kafkainternals.ConsumerGroup) {
+		c.OwnerReferences = append(c.OwnerReferences, metav1.OwnerReference{
+			APIVersion: "eventing.knative.dev/v1",
+			Kind:       "Trigger",
+			Name:       "test-trigger",
+			UID:        TriggerUUID,
+		})
+	}
+}