Repository: knative-sandbox/eventing-kafka-broker
Pull request #1820 (2022-01-27 22:16:58)

Adding Consumer Controller for for ConsumerGroup scaling
Signed-off-by: aavarghese <avarghese@us.ibm.com>

Part of #1537

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Adding new Consumer controller

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1820#" title="Author self-approved">aavarghese</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **julz** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @julz` 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":["julz"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/consumer/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1820/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1486825590282522624/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1820](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (4c9f7eb) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/db70212f7e886eb7be629b8b712b857a6be75a58?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (db70212) will **decrease** coverage by `10.77%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820/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/1820?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    #1820       +/-   ##
=============================================
- Coverage     68.79%   58.02%   -10.78%     
=============================================
  Files           131       65       -66     
  Lines          5492     3678     -1814     
  Branches        178        0      -178     
=============================================
- Hits           3778     2134     -1644     
+ Misses         1410     1333       -77     
+ Partials        304      211       -93     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

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/1820?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ontrol-plane/pkg/reconciler/consumer/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb250cm9sbGVyLmdv) | `100.00% <100.00%> (ø)` | |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820/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: |
| [...oker/dispatcher/main/ConsumerDeployerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lckRlcGxveWVyVmVydGljbGUuamF2YQ==) | | |
| [...roker/dispatcher/impl/filter/AttributesFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQXR0cmlidXRlc0ZpbHRlci5qYXZh) | | |
| [...atcher/impl/consumer/KafkaConsumerRecordUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9LYWZrYUNvbnN1bWVyUmVjb3JkVXRpbHMuamF2YQ==) | | |
| [...kafka/broker/dispatcher/impl/filter/AnyFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQW55RmlsdGVyLmphdmE=) | | |
| [...a/broker/dispatcher/impl/RecordDispatcherImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZWNvcmREaXNwYXRjaGVySW1wbC5qYXZh) | | |
| [...roker/dispatcher/impl/filter/BaseStringFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQmFzZVN0cmluZ0ZpbHRlci5qYXZh) | | |
| [...fka/broker/receiver/impl/handler/ProbeHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvUHJvYmVIYW5kbGVyLmphdmE=) | | |
| [...tive/eventing/kafka/broker/core/utils/Logging.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvTG9nZ2luZy5qYXZh) | | |
| ... and [63 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820?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/1820?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [db70212...4c9f7eb](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1820?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: @aavarghese: 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-upgrade-tests | 4c9f7ebee8ecb71e4a409a7553e862ae30401697 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1820/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1486825590223802368) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`

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

<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: @aavarghese: The following tests **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-upgrade-tests | 4c9f7ebee8ecb71e4a409a7553e862ae30401697 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1820/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1486825590223802368) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 4c9f7ebee8ecb71e4a409a7553e862ae30401697 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1820/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1486825590194442240) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<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 -->
aavarghese: /cc @pierDipi 
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/1820#" title="Author self-approved">aavarghese</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1820#pullrequestreview-866241923" title="LGTM">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)~~ [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":[]} -->
pierDipi: /test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests

Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index ec5cec4b4..600ae4aac 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -170,6 +170,7 @@ aliases:
   - nicolaferraro
   - rhuss
   knative-admin:
+  - carlisia
   - csantanapr
   - dprotaso
   - evankanderson
@@ -180,13 +181,13 @@ aliases:
   - knative-test-reporter-robot
   - lance
   - pmorie
-  - psschwei
   - rhuss
   - thisisnotapril
   - vaikas
+  - xtreme-sameer-vohra
   knative-release-leads:
-  - dprotaso
-  - psschwei
+  - carlisia
+  - xtreme-sameer-vohra
   knative-robots:
   - knative-prow-releaser-robot
   - knative-prow-robot
@@ -230,7 +231,9 @@ aliases:
   productivity-wg-leads:
   - chizhg
   productivity-writers:
+  - cardil
   - chizhg
+  - psschwei
   security-wg-leads:
   - evankanderson
   - julz
diff --git a/control-plane/pkg/reconciler/consumer/controller.go b/control-plane/pkg/reconciler/consumer/controller.go
new file mode 100644
index 000000000..7b7882fec
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumer/controller.go
@@ -0,0 +1,66 @@
+/*
+ * 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 consumer
+
+import (
+	"context"
+
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/resolver"
+
+	"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"
+	creconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumer"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
+	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
+)
+
+func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
+
+	consumerInformer := consumer.Get(ctx)
+
+	r := &Reconciler{
+		SerDe:               formatSerDeFromString(configs.DataPlaneConfigFormat),
+		ConsumerGroupLister: consumergroup.Get(ctx).Lister(),
+		SecretLister:        secretinformer.Get(ctx).Lister(),
+		PodLister:           podinformer.Get(ctx).Lister(),
+		KubeClient:          kubeclient.Get(ctx),
+	}
+
+	impl := creconciler.NewImpl(ctx, r)
+
+	r.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
+
+	consumerInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
+
+	r.Tracker = impl.Tracker
+	secretinformer.Get(ctx).Informer().AddEventHandler(controller.HandleAll(r.Tracker.OnChanged))
+
+	return impl
+}
+
+func formatSerDeFromString(val string) contract.FormatSerDe {
+	switch val {
+	case "protobuf":
+		return contract.FormatSerDe{Format: contract.Protobuf}
+	default:
+		return contract.FormatSerDe{Format: contract.Json}
+	}
+}
diff --git a/data-plane/config/sourcev2/500-dispatcher.yaml b/data-plane/config/sourcev2/500-dispatcher.yaml
new file mode 100644
index 000000000..770525ad2
--- /dev/null
+++ b/data-plane/config/sourcev2/500-dispatcher.yaml
@@ -0,0 +1,157 @@
+---
+
+# 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.
+
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: kafka-source-dispatcher
+  namespace: knative-eventing
+  labels:
+    app: kafka-source-dispatcher
+    kafka.eventing.knative.dev/release: devel
+spec:
+  serviceName: kafka-source-dispatcher
+  selector:
+    matchLabels:
+      app: kafka-source-dispatcher
+  template:
+    metadata:
+      name: kafka-source-dispatcher
+      labels:
+        app: kafka-source-dispatcher
+        kafka.eventing.knative.dev/release: devel
+    spec:
+      serviceAccountName: knative-kafka-source-data-plane
+      securityContext:
+        runAsNonRoot: true
+      containers:
+        - name: kafka-source-dispatcher
+          image: ${KNATIVE_KAFKA_DISPATCHER_IMAGE}
+          imagePullPolicy: IfNotPresent
+          volumeMounts:
+            - mountPath: /etc/config
+              name: config-kafka-source-data-plane
+              readOnly: true
+            - mountPath: /etc/contract-resources
+              name: contract-resources
+              readOnly: true
+            - mountPath: /tmp
+              name: cache
+            - mountPath: /etc/logging
+              name: kafka-config-logging
+              readOnly: true
+            - mountPath: /etc/tracing
+              name: config-tracing
+              readOnly: true
+          ports:
+            - containerPort: 9090
+              name: http-metrics
+              protocol: TCP
+          env:
+            - name: SERVICE_NAME
+              value: "kafka-source-dispatcher"
+            - name: SERVICE_NAMESPACE
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.namespace
+            - name: PRODUCER_CONFIG_FILE_PATH
+              value: /etc/config/config-kafka-source-producer.properties
+            - name: CONSUMER_CONFIG_FILE_PATH
+              value: /etc/config/config-kafka-source-consumer.properties
+            - name: WEBCLIENT_CONFIG_FILE_PATH
+              value: /etc/config/config-kafka-source-webclient.properties
+            - name: DATA_PLANE_CONFIG_FILE_PATH
+              value: /etc/contract-resources/data
+            - name: EGRESSES_INITIAL_CAPACITY
+              value: "20"
+            - name: INSTANCE_ID
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.uid
+            - name: METRICS_PATH
+              value: /metrics
+            - name: METRICS_PORT
+              value: "9090"
+            - name: METRICS_PUBLISH_QUANTILES
+              value: "false"
+            - name: METRICS_JVM_ENABLED
+              value: "false"
+            - name: CONFIG_TRACING_PATH
+              value: "/etc/tracing"
+            # https://github.com/fabric8io/kubernetes-client/issues/2212
+            - name: HTTP2_DISABLE
+              value: "true"
+            # This should be set according to initial delay seconds
+            - name: WAIT_STARTUP_SECONDS
+              value: "8"
+          command:
+            - "java"
+          args:
+            - "-Dlogback.configurationFile=/etc/logging/config.xml"
+            - "-jar"
+            - "/app/app.jar"
+
+          resources:
+            requests:
+              cpu: 1000m
+              # 600Mi for virtual replicas + 100Mi overhead
+              memory: 700Mi
+            limits:
+              cpu: 2000m
+              memory: 1000Mi
+
+          livenessProbe:
+            failureThreshold: 3
+            httpGet:
+              port: 9090
+              path: /metrics
+              scheme: HTTP
+            initialDelaySeconds: 10
+            periodSeconds: 3
+            successThreshold: 1
+            timeoutSeconds: 1
+          readinessProbe:
+            failureThreshold: 3
+            httpGet:
+              port: 9090
+              path: /metrics
+              scheme: HTTP
+            initialDelaySeconds: 10
+            periodSeconds: 3
+            successThreshold: 1
+            timeoutSeconds: 1
+          terminationMessagePolicy: FallbackToLogsOnError
+          terminationMessagePath: /dev/temination-log
+          securityContext:
+            allowPrivilegeEscalation: false
+            privileged: false
+            readOnlyRootFilesystem: true
+      volumes:
+        - name: config-kafka-source-data-plane
+          configMap:
+            name: config-kafka-source-data-plane
+        - name: cache
+          emptyDir: { }
+        - name: kafka-config-logging
+          configMap:
+            name: kafka-config-logging
+        - name: config-tracing
+          configMap:
+            name: config-tracing
+      restartPolicy: Always
+      dnsConfig:
+        options:
+          - name: single-request-reopen


Test patch:
diff --git a/control-plane/pkg/reconciler/consumer/controller_test.go b/control-plane/pkg/reconciler/consumer/controller_test.go
new file mode 100644
index 000000000..d2d8e77de
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumer/controller_test.go
@@ -0,0 +1,85 @@
+/*
+ * 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 consumer
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+
+	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/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"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+
+	_ "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)
+
+	dynamicScheme := runtime.NewScheme()
+	_ = fakekubeclientset.AddToScheme(dynamicScheme)
+
+	dynamicclient.With(ctx, dynamicScheme)
+
+	configs := &config.Env{
+		DataPlaneConfigFormat: "json",
+	}
+
+	controller := NewController(ctx, configs)
+	if controller == nil {
+		t.Error("failed to create controller: <nil>")
+	}
+}
+
+func TestFormatSerDeFromString(t *testing.T) {
+	tt := []struct {
+		format string
+		serde  contract.FormatSerDe
+	}{
+		{
+			format: "protobuf",
+			serde:  contract.FormatSerDe{Format: contract.Protobuf},
+		},
+		{
+			format: "json",
+			serde:  contract.FormatSerDe{Format: contract.Json},
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.format, func(t *testing.T) {
+
+			got := formatSerDeFromString(tc.format)
+
+			require.Equal(t, tc.serde, got, "%v %v", tc.serde, got)
+		})
+	}
+
+}