Repository: knative-sandbox/eventing-kafka-broker
Pull request #1709 (2022-01-07 20:02:08)

[data-plane]: Bump opentelemetry.version from 1.9.1 to 1.10.0 in /data-plane
Bumps `opentelemetry.version` from 1.9.1 to 1.10.0.
Updates `opentelemetry-bom` from 1.9.1 to 1.10.0
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/open-telemetry/opentelemetry-java/releases">opentelemetry-bom's releases</a>.</em></p>
<blockquote>
<h2>Version 1.10.0</h2>
<p>(These release notes contain the content of the 1.10 RCs)</p>
<p>With this release, the OpenTelemetry metrics API has been marked stable - you can use it through the usual <code>opentelemetry-api</code> artifact. The SDK is still under development and not yet stable, but we hope this can open the door to instrumentation libraries using the metrics API.</p>
<h3>API</h3>
<ul>
<li>Performance of <code>GlobalOpenTelemetry.get</code> improved</li>
<li><code>buildWithCallback</code> for asynchronous metrics now return interfaces instead of void. The interfaces are empty but will allow adding functionality in the future</li>
<li>BREAKING CHANGE: <code>Double/LongMeasurement.observe</code> have been removed</li>
<li>BREAKING CHANGE: <code>GlobalMeterProvider</code> has been removed</li>
<li>BREAKING CHANGE: <code>ObservableMeasurement</code>, an empty interface, has been removed. This type was not previously deprecated but is expected to have no use in apps due to the lack of functionality</li>
<li>The Metrics API has been merged into the <code>opentelemetry-api</code> artifact. <code>OpenTelemetry.getMeterProvider()</code> is the new entrypoint</li>
<li>BREAKING CHANGE: Bound metrics instruments have been removed for now to allow more time to bake while still providing a stable metrics API</li>
<li><code>Double/LongMeasurement.observe</code> has been renamed to <code>record</code>.<code>observe</code> is deprecated in this release</li>
<li><code>GlobalMeterProvider</code> has been deprecated. <code>GlobalOpenTelemetry.getMeterProvider</code> should be used instead</li>
<li>A warning is logged now when registering async instruments with the same name. Previously subsequent invocations were just ignored.</li>
<li><code>GlobalOpenTelemetry</code> extended with helpers for meter creation</li>
</ul>
<h3>SDK</h3>
<ul>
<li>The semantic conventions have been updated to 1.8.0</li>
<li>Deprecated methods have been removed from the <code>opentelemetry-sdk-autoconfigure</code> artifact.</li>
</ul>
<h4>Exporter</h4>
<ul>
<li>The OkHttp gRPC exporters now support experimental retry</li>
<li>OkHttp dispatcher threads are now spawned as daemon threads</li>
<li>The JPMS module name for the logs exporter has been fixed</li>
<li>Metrics exporters can have temporality configured</li>
<li>HTTP exporters now support experimental retry</li>
<li>Jaeger exporter allows setting trusted certificates</li>
<li>gRPC exporter metric typos corected</li>
</ul>
<h4>Metrics</h4>
<ul>
<li><code>InMemoryMetricExporter</code> has been moved to the <code>opentelemetry-sdk-metrics-testing</code> artifact. The current class has been deprecated in this release</li>
<li>Metric instrument usage violations consistently report issues as debug logs</li>
<li>Some user callbacks have been wrapped to catch exceptions instead of throwing</li>
<li>MinMaxSumCount/Count aggregation has been removed</li>
<li>Empty aggregator is renamed to <code>drop</code></li>
<li>Cumulative aggregations will not be aggressively dropped every collection cycle</li>
</ul>
<h4>Logs</h4>
<ul>
<li>A <code>opentelemetry-sdk-logs-testing</code> module has been added</li>
<li><code>SdkLogEmitterProvider</code> is now available through <code>OpenTelemetrySdk</code></li>
<li>LogDataBuilder can now take a SpanContext directly</li>
<li>SdkLogEmitterProvider.get convenience method added</li>
</ul>
<h4>AWS</h4>
<!-- raw HTML omitted -->
</blockquote>
<p>... (truncated)</p>
</details>
<details>
<summary>Changelog</summary>
<p><em>Sourced from <a href="https://github.com/open-telemetry/opentelemetry-java/blob/main/CHANGELOG.md">opentelemetry-bom's changelog</a>.</em></p>
<blockquote>
<h2>Version 1.10.0 (unreleased):</h2>
<h2>Version 1.10.0 RC2 2021-12-23:</h2>
<h3>API</h3>
<ul>
<li>Performance of <code>GlobalOpenTelemetry.get</code> improved</li>
<li><code>buildWithCallback</code> for asynchronous metrics now return interfaces instead of void. The interfaces are empty but will allow adding functionality in the future</li>
<li>BREAKING CHANGE: <code>Double/LongMeasurement.observe</code> have been removed</li>
<li>BREAKING CHANGE: <code>GlobalMeterProvider</code> has been removed</li>
<li>BREAKING CHANGE: <code>ObservableMeasurement</code>, an empty interface, has been removed. This type was not previously deprecated but is expected to have no use in apps due to the lack of functionality</li>
</ul>
<h3>SDK</h3>
<ul>
<li>HTTP exporters now support experimental retry</li>
<li>Deprecated methods have been removed from the <code>opentelemetry-sdk-autoconfigure</code> artifact.</li>
</ul>
<h4>Metrics</h4>
<ul>
<li><code>InMemoryMetricExporter</code> has been moved to the <code>opentelemetry-sdk-metrics-testing</code> artifact. The current class has been deprecated in this release</li>
<li>Metric instrument usage violations consistently report issues as debug logs</li>
</ul>
<h4>Logs</h4>
<ul>
<li>A <code>opentelemetry-sdk-logs-testing</code> module has been added</li>
<li><code>SdkLogEmitterProvider</code> is now available through <code>OpenTelemetrySdk</code></li>
</ul>
<h4>AWS</h4>
<ul>
<li>HTTP requests now use OkHttp instead of the JDK</li>
</ul>
<h3>OpenCensus Shim</h3>
<ul>
<li>Shim span attributes are set before the span is created instead of after</li>
<li>Exceptions are not thrown when activating a null span</li>
</ul>
<h2>Version 1.10.0 RC1 2021-12-10:</h2>
<h3>API</h3>
<ul>
<li>The Metrics API has been merged into the <code>opentelemetry-api</code> artifact. <code>OpenTelemetry.getMeterProvider()</code> is the new entrypoint</li>
<li>BREAKING CHANGE: Bound metrics instruments have been removed for now to allow more time to bake while still providing a stable metrics API</li>
<li><code>Double/LongMeasurement.observe</code> has been renamed to <code>record</code>.<code>observe</code> is deprecated in this release</li>
<li><code>GlobalMeterProvider</code> has been deprecated. <code>GlobalOpenTelemetry.getMeterProvider</code> should be used instead</li>
</ul>
<h3>SDK</h3>
<ul>
<li>The semantic conventions have been updated to 1.8.0</li>
</ul>
<h4>Exporter</h4>
<ul>
<li>The OkHttp gRPC exporters now support experimental retry</li>
<li>OkHttp dispatcher threads are now spawned as daemon threads</li>
</ul>
<!-- raw HTML omitted -->
</blockquote>
<p>... (truncated)</p>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/89675582a173dc2df2aa43040decdb2a43ee0989"><code>8967558</code></a> Use custom resource to avoid version number in assertion string. (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4065">#4065</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/edbe46a9f7a230f28fc53497eeead05ce084f18f"><code>edbe46a</code></a> Expose setTrustedCertificates in JaegerGrpcSpanExporterBuilder for mTLS Suppo...</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/c28797ae89110ade7bd1bc6a3d162abd3cebda36"><code>c28797a</code></a> Use configured classloader for resource SPI. (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4064">#4064</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/55bb83d2c1eedc405539403e7c01ded5074f74bc"><code>55bb83d</code></a> Log warning when duplicate async instruments are registered (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4020">#4020</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/429afa1c991f2162cf4aa3e1e8c33ac8e580e825"><code>429afa1</code></a> Add toString to SDK trace components and print autoconfigured SDK (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4047">#4047</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/96b7895173e8a30c319f44b38a68523d5d95fffe"><code>96b7895</code></a> Add DoubleHistogramPointData assertion: hasSumGreaterThan() (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4050">#4050</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/dfb4a74ac5a61646c1a6b0abe9d1439c369441be"><code>dfb4a74</code></a> Update errorprone (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4048">#4048</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/10f5efa14da30b4b6eb124a5c938f50904406186"><code>10f5efa</code></a> Add log autoconfigure support (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4015">#4015</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/f5a7cd4741f307382753ae098988634a718fbabc"><code>f5a7cd4</code></a> Update dependencies (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4045">#4045</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/e65567d7ac09e0f739f60ac4e54961eb031cbf7c"><code>e65567d</code></a> OTLP Marshalers for Exponential Histogram (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/3995">#3995</a>)</li>
<li>Additional commits viewable in <a href="https://github.com/open-telemetry/opentelemetry-java/compare/v1.9.1...v1.10.0">compare view</a></li>
</ul>
</details>
<br />

Updates `opentelemetry-extension-trace-propagators` from 1.9.1 to 1.10.0
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/open-telemetry/opentelemetry-java/releases">opentelemetry-extension-trace-propagators's releases</a>.</em></p>
<blockquote>
<h2>Version 1.10.0</h2>
<p>(These release notes contain the content of the 1.10 RCs)</p>
<p>With this release, the OpenTelemetry metrics API has been marked stable - you can use it through the usual <code>opentelemetry-api</code> artifact. The SDK is still under development and not yet stable, but we hope this can open the door to instrumentation libraries using the metrics API.</p>
<h3>API</h3>
<ul>
<li>Performance of <code>GlobalOpenTelemetry.get</code> improved</li>
<li><code>buildWithCallback</code> for asynchronous metrics now return interfaces instead of void. The interfaces are empty but will allow adding functionality in the future</li>
<li>BREAKING CHANGE: <code>Double/LongMeasurement.observe</code> have been removed</li>
<li>BREAKING CHANGE: <code>GlobalMeterProvider</code> has been removed</li>
<li>BREAKING CHANGE: <code>ObservableMeasurement</code>, an empty interface, has been removed. This type was not previously deprecated but is expected to have no use in apps due to the lack of functionality</li>
<li>The Metrics API has been merged into the <code>opentelemetry-api</code> artifact. <code>OpenTelemetry.getMeterProvider()</code> is the new entrypoint</li>
<li>BREAKING CHANGE: Bound metrics instruments have been removed for now to allow more time to bake while still providing a stable metrics API</li>
<li><code>Double/LongMeasurement.observe</code> has been renamed to <code>record</code>.<code>observe</code> is deprecated in this release</li>
<li><code>GlobalMeterProvider</code> has been deprecated. <code>GlobalOpenTelemetry.getMeterProvider</code> should be used instead</li>
<li>A warning is logged now when registering async instruments with the same name. Previously subsequent invocations were just ignored.</li>
<li><code>GlobalOpenTelemetry</code> extended with helpers for meter creation</li>
</ul>
<h3>SDK</h3>
<ul>
<li>The semantic conventions have been updated to 1.8.0</li>
<li>Deprecated methods have been removed from the <code>opentelemetry-sdk-autoconfigure</code> artifact.</li>
</ul>
<h4>Exporter</h4>
<ul>
<li>The OkHttp gRPC exporters now support experimental retry</li>
<li>OkHttp dispatcher threads are now spawned as daemon threads</li>
<li>The JPMS module name for the logs exporter has been fixed</li>
<li>Metrics exporters can have temporality configured</li>
<li>HTTP exporters now support experimental retry</li>
<li>Jaeger exporter allows setting trusted certificates</li>
<li>gRPC exporter metric typos corected</li>
</ul>
<h4>Metrics</h4>
<ul>
<li><code>InMemoryMetricExporter</code> has been moved to the <code>opentelemetry-sdk-metrics-testing</code> artifact. The current class has been deprecated in this release</li>
<li>Metric instrument usage violations consistently report issues as debug logs</li>
<li>Some user callbacks have been wrapped to catch exceptions instead of throwing</li>
<li>MinMaxSumCount/Count aggregation has been removed</li>
<li>Empty aggregator is renamed to <code>drop</code></li>
<li>Cumulative aggregations will not be aggressively dropped every collection cycle</li>
</ul>
<h4>Logs</h4>
<ul>
<li>A <code>opentelemetry-sdk-logs-testing</code> module has been added</li>
<li><code>SdkLogEmitterProvider</code> is now available through <code>OpenTelemetrySdk</code></li>
<li>LogDataBuilder can now take a SpanContext directly</li>
<li>SdkLogEmitterProvider.get convenience method added</li>
</ul>
<h4>AWS</h4>
<!-- raw HTML omitted -->
</blockquote>
<p>... (truncated)</p>
</details>
<details>
<summary>Changelog</summary>
<p><em>Sourced from <a href="https://github.com/open-telemetry/opentelemetry-java/blob/main/CHANGELOG.md">opentelemetry-extension-trace-propagators's changelog</a>.</em></p>
<blockquote>
<h2>Version 1.10.0 (unreleased):</h2>
<h2>Version 1.10.0 RC2 2021-12-23:</h2>
<h3>API</h3>
<ul>
<li>Performance of <code>GlobalOpenTelemetry.get</code> improved</li>
<li><code>buildWithCallback</code> for asynchronous metrics now return interfaces instead of void. The interfaces are empty but will allow adding functionality in the future</li>
<li>BREAKING CHANGE: <code>Double/LongMeasurement.observe</code> have been removed</li>
<li>BREAKING CHANGE: <code>GlobalMeterProvider</code> has been removed</li>
<li>BREAKING CHANGE: <code>ObservableMeasurement</code>, an empty interface, has been removed. This type was not previously deprecated but is expected to have no use in apps due to the lack of functionality</li>
</ul>
<h3>SDK</h3>
<ul>
<li>HTTP exporters now support experimental retry</li>
<li>Deprecated methods have been removed from the <code>opentelemetry-sdk-autoconfigure</code> artifact.</li>
</ul>
<h4>Metrics</h4>
<ul>
<li><code>InMemoryMetricExporter</code> has been moved to the <code>opentelemetry-sdk-metrics-testing</code> artifact. The current class has been deprecated in this release</li>
<li>Metric instrument usage violations consistently report issues as debug logs</li>
</ul>
<h4>Logs</h4>
<ul>
<li>A <code>opentelemetry-sdk-logs-testing</code> module has been added</li>
<li><code>SdkLogEmitterProvider</code> is now available through <code>OpenTelemetrySdk</code></li>
</ul>
<h4>AWS</h4>
<ul>
<li>HTTP requests now use OkHttp instead of the JDK</li>
</ul>
<h3>OpenCensus Shim</h3>
<ul>
<li>Shim span attributes are set before the span is created instead of after</li>
<li>Exceptions are not thrown when activating a null span</li>
</ul>
<h2>Version 1.10.0 RC1 2021-12-10:</h2>
<h3>API</h3>
<ul>
<li>The Metrics API has been merged into the <code>opentelemetry-api</code> artifact. <code>OpenTelemetry.getMeterProvider()</code> is the new entrypoint</li>
<li>BREAKING CHANGE: Bound metrics instruments have been removed for now to allow more time to bake while still providing a stable metrics API</li>
<li><code>Double/LongMeasurement.observe</code> has been renamed to <code>record</code>.<code>observe</code> is deprecated in this release</li>
<li><code>GlobalMeterProvider</code> has been deprecated. <code>GlobalOpenTelemetry.getMeterProvider</code> should be used instead</li>
</ul>
<h3>SDK</h3>
<ul>
<li>The semantic conventions have been updated to 1.8.0</li>
</ul>
<h4>Exporter</h4>
<ul>
<li>The OkHttp gRPC exporters now support experimental retry</li>
<li>OkHttp dispatcher threads are now spawned as daemon threads</li>
</ul>
<!-- raw HTML omitted -->
</blockquote>
<p>... (truncated)</p>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/89675582a173dc2df2aa43040decdb2a43ee0989"><code>8967558</code></a> Use custom resource to avoid version number in assertion string. (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4065">#4065</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/edbe46a9f7a230f28fc53497eeead05ce084f18f"><code>edbe46a</code></a> Expose setTrustedCertificates in JaegerGrpcSpanExporterBuilder for mTLS Suppo...</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/c28797ae89110ade7bd1bc6a3d162abd3cebda36"><code>c28797a</code></a> Use configured classloader for resource SPI. (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4064">#4064</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/55bb83d2c1eedc405539403e7c01ded5074f74bc"><code>55bb83d</code></a> Log warning when duplicate async instruments are registered (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4020">#4020</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/429afa1c991f2162cf4aa3e1e8c33ac8e580e825"><code>429afa1</code></a> Add toString to SDK trace components and print autoconfigured SDK (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4047">#4047</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/96b7895173e8a30c319f44b38a68523d5d95fffe"><code>96b7895</code></a> Add DoubleHistogramPointData assertion: hasSumGreaterThan() (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4050">#4050</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/dfb4a74ac5a61646c1a6b0abe9d1439c369441be"><code>dfb4a74</code></a> Update errorprone (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4048">#4048</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/10f5efa14da30b4b6eb124a5c938f50904406186"><code>10f5efa</code></a> Add log autoconfigure support (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4015">#4015</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/f5a7cd4741f307382753ae098988634a718fbabc"><code>f5a7cd4</code></a> Update dependencies (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/4045">#4045</a>)</li>
<li><a href="https://github.com/open-telemetry/opentelemetry-java/commit/e65567d7ac09e0f739f60ac4e54961eb031cbf7c"><code>e65567d</code></a> OTLP Marshalers for Exponential Histogram (<a href="https://github-redirect.dependabot.com/open-telemetry/opentelemetry-java/issues/3995">#3995</a>)</li>
<li>Additional commits viewable in <a href="https://github.com/open-telemetry/opentelemetry-java/compare/v1.9.1...v1.10.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/1709#" 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 **psschwei** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @psschwei` 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":["psschwei"]} -->
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/1709#" 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/1709?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1709](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1709?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (9778f0c) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/4f388658cb55b35d069e33bac496d940e1d42db6?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (4f38865) will **decrease** coverage by `0.04%`.
> The diff coverage is `n/a`.

> :exclamation: Current head 9778f0c differs from pull request most recent head ff92007. Consider uploading reports for the commit ff92007 to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1709/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/1709?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    #1709      +/-   ##
============================================
- Coverage     72.81%   72.76%   -0.05%     
+ Complexity      597      595       -2     
============================================
  Files           120      120              
  Lines          4443     4443              
  Branches        171      171              
============================================
- Hits           3235     3233       -2     
- Misses          944      947       +3     
+ Partials        264      263       -1     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `81.80% <ø> (-0.11%)` | :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/1709?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...dispatcher/impl/consumer/BaseConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1709/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9CYXNlQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | `87.50% <0.00%> (-6.25%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1709?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/1709?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [4f38865...ff92007](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1709?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).


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 8c9f8b626..c08ad5e87 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
@@ -93,6 +93,12 @@ type ConsumerGroupStatus struct {
 	// DeliveryStatus contains a resolved URL to the dead letter sink address, and any other
 	// resolved delivery options.
 	eventingduckv1.DeliveryStatus `json:",inline"`
+
+	// Replicas is the latest observed number of replicas of the given Template.
+	// These are replicas in the sense that they are instantiations of the
+	// same Template, but individual replicas also have a consistent identity.
+	// +optional
+	Replicas *int32 `json:"replicas,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
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 93284aba9..cc293e0de 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
@@ -19,10 +19,12 @@ package v1alpha1
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	bindings "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
+	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 )
 
@@ -59,7 +61,9 @@ type ConsumerSpec struct {
 	// More info: https://kafka.apache.org/documentation/#consumerconfigs
 	Configs ConsumerConfigs `json:"configs,omitempty"`
 
-	// TODO Add auth
+	// Auth is the auth configuration for the Consumer.
+	// +optional
+	Auth *Auth `json:"auth,omitempty"`
 
 	// DeliverySpec contains the delivery options for event senders.
 	// +optional
@@ -71,6 +75,16 @@ type ConsumerSpec struct {
 
 	// Subscriber is the addressable that receives events that pass the Filters.
 	Subscriber duckv1.Destination `json:"subscriber"`
+
+	// CloudEventOverrides defines overrides to control the output format and
+	// modifications of the event sent to the subscriber.
+	// +optional
+	CloudEventOverrides *duckv1.CloudEventOverrides `json:"ceOverrides,omitempty"`
+}
+
+type Auth struct {
+	NetSpec  *bindings.KafkaNetSpec
+	AuthSpec *eventingv1alpha1.Auth
 }
 
 type DeliverySpec struct {
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
index d2d8c9a78..60024489f 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
@@ -23,11 +23,40 @@ package v1alpha1
 
 import (
 	runtime "k8s.io/apimachinery/pkg/runtime"
-	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+	v1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	apis "knative.dev/pkg/apis"
+	v1 "knative.dev/pkg/apis/duck/v1"
 )
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Auth) DeepCopyInto(out *Auth) {
+	*out = *in
+	if in.NetSpec != nil {
+		in, out := &in.NetSpec, &out.NetSpec
+		*out = new(v1beta1.KafkaNetSpec)
+		(*in).DeepCopyInto(*out)
+	}
+	if in.AuthSpec != nil {
+		in, out := &in.AuthSpec, &out.AuthSpec
+		*out = new(eventingv1alpha1.Auth)
+		(*in).DeepCopyInto(*out)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Auth.
+func (in *Auth) DeepCopy() *Auth {
+	if in == nil {
+		return nil
+	}
+	out := new(Auth)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Consumer) DeepCopyInto(out *Consumer) {
 	*out = *in
@@ -180,6 +209,11 @@ func (in *ConsumerGroupStatus) DeepCopyInto(out *ConsumerGroupStatus) {
 		(*in).DeepCopyInto(*out)
 	}
 	in.DeliveryStatus.DeepCopyInto(&out.DeliveryStatus)
+	if in.Replicas != nil {
+		in, out := &in.Replicas, &out.Replicas
+		*out = new(int32)
+		**out = **in
+	}
 	return
 }
 
@@ -235,6 +269,11 @@ func (in *ConsumerSpec) DeepCopyInto(out *ConsumerSpec) {
 		copy(*out, *in)
 	}
 	in.Configs.DeepCopyInto(&out.Configs)
+	if in.Auth != nil {
+		in, out := &in.Auth, &out.Auth
+		*out = new(Auth)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.Delivery != nil {
 		in, out := &in.Delivery, &out.Delivery
 		*out = new(DeliverySpec)
@@ -246,6 +285,11 @@ func (in *ConsumerSpec) DeepCopyInto(out *ConsumerSpec) {
 		(*in).DeepCopyInto(*out)
 	}
 	in.Subscriber.DeepCopyInto(&out.Subscriber)
+	if in.CloudEventOverrides != nil {
+		in, out := &in.CloudEventOverrides, &out.CloudEventOverrides
+		*out = new(v1.CloudEventOverrides)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -305,7 +349,7 @@ func (in *DeliverySpec) DeepCopyInto(out *DeliverySpec) {
 	*out = *in
 	if in.DeliverySpec != nil {
 		in, out := &in.DeliverySpec, &out.DeliverySpec
-		*out = new(v1.DeliverySpec)
+		*out = new(duckv1.DeliverySpec)
 		(*in).DeepCopyInto(*out)
 	}
 	return
diff --git a/control-plane/pkg/reconciler/consumergroup/filter.go b/control-plane/pkg/reconciler/consumergroup/filter.go
index 1ce2b2a2f..0ff1b0927 100644
--- a/control-plane/pkg/reconciler/consumergroup/filter.go
+++ b/control-plane/pkg/reconciler/consumergroup/filter.go
@@ -19,6 +19,8 @@ package consumergroup
 import (
 	"strings"
 
+	"k8s.io/apimachinery/pkg/types"
+
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 )
 
@@ -41,3 +43,20 @@ func Filter(userFacingResource string) func(obj interface{}) bool {
 		return false
 	}
 }
+
+// Enqueue enqueues using the provided enqueue function the resource associated with a ConsumerGroup
+func Enqueue(userFacingResource string, enqueue func(key types.NamespacedName)) func(obj interface{}) {
+	userFacingResource = strings.ToLower(userFacingResource)
+	return func(obj interface{}) {
+		cg, ok := obj.(*kafkainternals.ConsumerGroup)
+		if !ok {
+			return
+		}
+
+		for _, or := range cg.OwnerReferences {
+			if strings.ToLower(or.Kind) == userFacingResource {
+				enqueue(types.NamespacedName{Namespace: cg.GetNamespace(), Name: or.Name})
+			}
+		}
+	}
+}
diff --git a/control-plane/pkg/reconciler/source/v2/controllerv2.go b/control-plane/pkg/reconciler/source/v2/controllerv2.go
new file mode 100644
index 000000000..dbd161ca1
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/v2/controllerv2.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 v2
+
+import (
+	"context"
+
+	"k8s.io/client-go/tools/cache"
+	consumergroupclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
+	consumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup"
+	"knative.dev/pkg/controller"
+
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+
+	kafkainformer "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource"
+	"knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumergroup"
+)
+
+func NewControllerV2(ctx context.Context, configs *config.Env) *controller.Impl {
+
+	kafkaInformer := kafkainformer.Get(ctx)
+	consumerGroupInformer := consumergroupinformer.Get(ctx)
+
+	sources.RegisterAlternateKafkaConditionSet(conditionSet)
+
+	r := &ReconcilerV2{
+		ConsumerGroupLister: consumerGroupInformer.Lister(),
+		InternalsClient:     consumergroupclient.Get(ctx),
+	}
+
+	impl := kafkasource.NewImpl(ctx, r)
+
+	kafkaInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		Handler: controller.HandleAll(impl.Enqueue),
+	})
+
+	// ConsumerGroup changes and enqueue associated KafkaSource
+	consumerGroupInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: consumergroup.Filter("kafkasource"),
+		Handler:    controller.HandleAll(consumergroup.Enqueue("kafkasource", impl.EnqueueKey)),
+	})
+	return impl
+}
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2.go b/control-plane/pkg/reconciler/source/v2/sourcev2.go
new file mode 100644
index 000000000..9886d206f
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2.go
@@ -0,0 +1,137 @@
+/*
+ * 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 v2
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/api/equality"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/kmeta"
+	"knative.dev/pkg/reconciler"
+
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	internalscg "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/clientset/versioned"
+	internalslst "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
+)
+
+const (
+	DefaultDeliveryOrder = internals.Ordered
+
+	KafkaConditionConsumerGroup apis.ConditionType = "ConsumerGroup" //condition is registered by controller
+)
+
+var (
+	conditionSet = apis.NewLivingConditionSet(
+		KafkaConditionConsumerGroup,
+	)
+)
+
+type ReconcilerV2 struct {
+	ConsumerGroupLister internalslst.ConsumerGroupLister
+	InternalsClient     internalsclient.Interface
+}
+
+func (r *ReconcilerV2) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
+
+	consgroup, err := r.reconcileConsumerGroup(ctx, ks)
+	if err != nil {
+		ks.GetConditionSet().Manage(&ks.Status).MarkFalse(KafkaConditionConsumerGroup, "failed to reconcile consumer group", err.Error())
+		return err
+	}
+	ks.GetConditionSet().Manage(&ks.Status).MarkTrue(KafkaConditionConsumerGroup)
+
+	ks.Status.MarkSink(consgroup.Status.SubscriberURI)
+	ks.Status.Placeable = consgroup.Status.Placeable
+	ks.Status.Consumers = *consgroup.Status.Replicas
+
+	return nil
+}
+
+func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, ks *sources.KafkaSource) (*internalscg.ConsumerGroup, error) {
+
+	newcg := &internalscg.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      string(ks.UID),
+			Namespace: ks.Namespace,
+			OwnerReferences: []metav1.OwnerReference{
+				*kmeta.NewControllerRef(ks),
+			},
+		},
+		Spec: internalscg.ConsumerGroupSpec{
+			Template: internalscg.ConsumerTemplateSpec{
+				Spec: internalscg.ConsumerSpec{
+					Topics: ks.Spec.Topics,
+					Configs: internalscg.ConsumerConfigs{Configs: map[string]string{
+						"group.id":          ks.Spec.ConsumerGroup,
+						"bootstrap.servers": strings.Join(ks.Spec.BootstrapServers, ","),
+					}},
+					Auth: &internalscg.Auth{
+						NetSpec: &ks.Spec.KafkaAuthSpec.Net,
+					},
+					Delivery: &internalscg.DeliverySpec{
+						Ordering: DefaultDeliveryOrder,
+					},
+					Subscriber: ks.Spec.Sink,
+					CloudEventOverrides: &duckv1.CloudEventOverrides{
+						Extensions: ks.Spec.CloudEventOverrides.Extensions,
+					},
+				},
+			},
+			Replicas: ks.Spec.Consumers,
+		},
+	}
+
+	cg, err := r.ConsumerGroupLister.ConsumerGroups(ks.GetNamespace()).Get(string(ks.UID)) //Get by consumer group id
+	if err != nil && !apierrors.IsNotFound(err) {
+		return nil, err
+	}
+
+	if apierrors.IsNotFound(err) {
+		cg, err := r.InternalsClient.InternalV1alpha1().ConsumerGroups(newcg.GetNamespace()).Create(ctx, newcg, metav1.CreateOptions{})
+		if err != nil && !apierrors.IsAlreadyExists(err) {
+			return nil, fmt.Errorf("failed to create consumer group %s/%s: %w", newcg.GetNamespace(), newcg.GetName(), err)
+		}
+		if apierrors.IsAlreadyExists(err) {
+			return newcg, nil
+		}
+		return cg, nil
+	}
+
+	if equality.Semantic.DeepDerivative(newcg.Spec, cg.Spec) {
+		return cg, nil
+	}
+
+	newCg := &internalscg.ConsumerGroup{
+		TypeMeta:   cg.TypeMeta,
+		ObjectMeta: cg.ObjectMeta,
+		Spec:       newcg.Spec,
+		Status:     cg.Status,
+	}
+	if cg, err = r.InternalsClient.InternalV1alpha1().ConsumerGroups(cg.GetNamespace()).Update(ctx, newCg, metav1.UpdateOptions{}); err != nil {
+		return nil, fmt.Errorf("failed to update consumer group %s/%s: %w", newCg.GetNamespace(), newCg.GetName(), err)
+	}
+
+	return cg, nil
+}
diff --git a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
index 8f0ec27df..22229f0b9 100644
--- a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
@@ -88,7 +88,7 @@ func NewControllerV2(ctx context.Context, configs *config.Env) *controller.Impl
 	// ConsumerGroup changes and enqueue associated Trigger
 	consumerGroupInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: consumergroup.Filter("trigger"),
-		Handler:    controller.HandleAll(impl.Enqueue),
+		Handler:    controller.HandleAll(consumergroup.Enqueue("trigger", impl.EnqueueKey)),
 	})
 
 	return impl
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
index c1ef22033..8a79670b9 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
@@ -111,7 +111,6 @@ func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, trigger *event
 		}
 	}
 
-	//todo additionally handle trigger.Spec.Filter
 	newcg := &internalscg.ConsumerGroup{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      string(trigger.UID),
@@ -138,8 +137,6 @@ func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, trigger *event
 		},
 	}
 
-	//todo validate newcg
-
 	cg, err := r.ConsumerGroupLister.ConsumerGroups(trigger.GetNamespace()).Get(string(trigger.UID)) //Get by consumer group name
 	if err != nil && !apierrors.IsNotFound(err) {
 		return nil, err
@@ -166,7 +163,7 @@ func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, trigger *event
 		Spec:       newcg.Spec,
 		Status:     cg.Status,
 	}
-	if _, err := r.InternalsClient.InternalV1alpha1().ConsumerGroups(cg.GetNamespace()).Update(ctx, newCg, metav1.UpdateOptions{}); err != nil {
+	if cg, err = r.InternalsClient.InternalV1alpha1().ConsumerGroups(cg.GetNamespace()).Update(ctx, newCg, metav1.UpdateOptions{}); err != nil {
 		return nil, fmt.Errorf("failed to update consumer group %s/%s: %w", newCg.GetNamespace(), newCg.GetName(), err)
 	}
 
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index c7da64c13..cd4f96acb 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -1,5 +1,5 @@
 
-Lists of 159 third-party dependencies.
+Lists of 158 third-party dependencies.
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Classic Module (ch.qos.logback:logback-classic:1.2.10 - http://logback.qos.ch/logback-classic)
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Core Module (ch.qos.logback:logback-core:1.2.10 - http://logback.qos.ch/logback-core)
      (The Apache Software License, Version 2.0) Jackson-annotations (com.fasterxml.jackson.core:jackson-annotations:2.13.1 - http://github.com/FasterXML/jackson)
@@ -85,19 +85,18 @@ Lists of 159 third-party dependencies.
      (Apache License, Version 2.0) Netty/Transport/Classes/Epoll (io.netty:netty-transport-classes-epoll:4.1.72.Final - https://netty.io/netty-transport-classes-epoll/)
      (Apache License, Version 2.0) Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.72.Final - https://netty.io/netty-transport-native-epoll/)
      (Apache License, Version 2.0) Netty/Transport/Native/Unix/Common (io.netty:netty-transport-native-unix-common:4.1.72.Final - https://netty.io/netty-transport-native-unix-common/)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api-metrics:1.9.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-zipkin:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-extension-trace-propagators:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-common:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-logs:1.9.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-metrics:1.9.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-testing:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-trace:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-semconv:1.9.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-zipkin:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-extension-trace-propagators:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-common:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-logs:1.10.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-metrics:1.10.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-testing:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-trace:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-semconv:1.10.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient (io.prometheus:simpleclient:0.12.0 - http://github.com/prometheus/client_java/simpleclient)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient Common (io.prometheus:simpleclient_common:0.12.0 - http://github.com/prometheus/client_java/simpleclient_common)
      (The Apache Software License, Version 2.0) Prometheus Java Span Context Supplier - Common (io.prometheus:simpleclient_tracer_common:0.12.0 - http://github.com/prometheus/client_java/simpleclient_tracer/simpleclient_tracer_common)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 8463b365f..8fb9aa685 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -44,7 +44,7 @@
     <vertx.version>4.2.3</vertx.version>
     <cloudevents.sdk.version>2.3.0</cloudevents.sdk.version>
     <micrometer.version>1.8.1</micrometer.version>
-    <opentelemetry.version>1.9.1</opentelemetry.version>
+    <opentelemetry.version>1.10.0</opentelemetry.version>
     <jackson.version>2.13.1</jackson.version>
     <protobuf.version>3.19.2</protobuf.version>
     <slf4j.version>1.7.32</slf4j.version>


Test patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/filter_test.go b/control-plane/pkg/reconciler/consumergroup/filter_test.go
index e56dc59b3..5f00b23c7 100644
--- a/control-plane/pkg/reconciler/consumergroup/filter_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/filter_test.go
@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 )
@@ -76,3 +77,61 @@ func TestFilter(t *testing.T) {
 		})
 	}
 }
+
+func TestEnqueue(t *testing.T) {
+	tests := []struct {
+		name               string
+		resource           interface{}
+		userFacingResource string
+		want               int
+	}{
+		{
+			name:               "unknown type",
+			resource:           &kafkainternals.Consumer{},
+			userFacingResource: "trigger",
+			want:               0,
+		},
+		{
+			name: "trigger pass",
+			resource: &kafkainternals.ConsumerGroup{
+				TypeMeta: metav1.TypeMeta{},
+				ObjectMeta: metav1.ObjectMeta{
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							Kind: "trigger",
+						},
+					},
+				},
+			},
+			userFacingResource: "trigger",
+			want:               1,
+		},
+		{
+			name: "no pass",
+			resource: &kafkainternals.ConsumerGroup{
+				TypeMeta: metav1.TypeMeta{},
+				ObjectMeta: metav1.ObjectMeta{
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							Kind: "kafkasource",
+						},
+					},
+				},
+			},
+			userFacingResource: "trigger",
+			want:               0,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			called := 0
+			f := func(key types.NamespacedName) {
+				called++
+			}
+			Enqueue(tt.userFacingResource, f)(tt.resource)
+			if called != tt.want {
+				t.Errorf("Enqueue called %v, want %v", called, tt.want)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/source/v2/controllerv2_test.go b/control-plane/pkg/reconciler/source/v2/controllerv2_test.go
new file mode 100644
index 000000000..e378c42b1
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/v2/controllerv2_test.go
@@ -0,0 +1,68 @@
+/*
+ * 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 v2
+
+import (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake"
+	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
+	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
+	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+)
+
+func TestNewControllerV2(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	configs := &config.Env{
+		SystemNamespace:      "cm",
+		GeneralConfigMapName: "cm",
+	}
+
+	ctx, _ = fakekubeclient.With(
+		ctx,
+		&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      configs.GeneralConfigMapName,
+				Namespace: configs.SystemNamespace,
+			},
+		},
+	)
+	dynamicScheme := runtime.NewScheme()
+	_ = fakekubeclientset.AddToScheme(dynamicScheme)
+
+	dynamicclient.With(ctx, dynamicScheme)
+
+	controller := NewControllerV2(ctx, configs)
+	if controller == nil {
+		t.Error("failed to create controller: <nil>")
+	}
+}