Repository: knative-sandbox/eventing-kafka-broker
Pull request #1800 (2022-01-25 20:02:28)

[data-plane]: Bump slf4j.version from 1.7.33 to 1.7.35 in /data-plane
Bumps `slf4j.version` from 1.7.33 to 1.7.35.
Updates `slf4j-api` from 1.7.33 to 1.7.35
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/qos-ch/slf4j/commit/02860b67ef7ff39fa9c7d98fd00da2ee913faeda"><code>02860b6</code></a> prepare relase 1.7.35</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/a622f5186a57188dab7f71651245eb91c6ac263b"><code>a622f51</code></a> fix maven deploy issues</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/26068bd4bf93fcbd00185ad986dc43b79aceeb4a"><code>26068bd</code></a> slf4j no longer references log4j</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/0a21ee1ac1daa2d8e077bec68815421dd7a7a54a"><code>0a21ee1</code></a> replace references to slf4j-log4j12</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/51b6d20b71de75f69ee68167afbf4073c1be7c31"><code>51b6d20</code></a> prepare release 1.7.34</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/d22943faedd5da8d0321cf60437796fb53618481"><code>d22943f</code></a> relocate slf4j-log4j12 as slf4j-reload4j</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/19e36ffdca0218797cd23048b6547865e30e1d3a"><code>19e36ff</code></a> make VersionUtil more robust</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/d32d0535f7274a679c47d3354411476a86f5971a"><code>d32d053</code></a> fix SLF4J-535</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/2b657bf5dc575f32791648fd95260e33aa07687c"><code>2b657bf</code></a> start work on 1.7.33-SNAPSHOT</li>
<li>See full diff in <a href="https://github.com/qos-ch/slf4j/compare/v_1.7.33...v_1.7.35">compare view</a></li>
</ul>
</details>
<br />

Updates `slf4j-nop` from 1.7.33 to 1.7.35
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/qos-ch/slf4j/commit/02860b67ef7ff39fa9c7d98fd00da2ee913faeda"><code>02860b6</code></a> prepare relase 1.7.35</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/a622f5186a57188dab7f71651245eb91c6ac263b"><code>a622f51</code></a> fix maven deploy issues</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/26068bd4bf93fcbd00185ad986dc43b79aceeb4a"><code>26068bd</code></a> slf4j no longer references log4j</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/0a21ee1ac1daa2d8e077bec68815421dd7a7a54a"><code>0a21ee1</code></a> replace references to slf4j-log4j12</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/51b6d20b71de75f69ee68167afbf4073c1be7c31"><code>51b6d20</code></a> prepare release 1.7.34</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/d22943faedd5da8d0321cf60437796fb53618481"><code>d22943f</code></a> relocate slf4j-log4j12 as slf4j-reload4j</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/19e36ffdca0218797cd23048b6547865e30e1d3a"><code>19e36ff</code></a> make VersionUtil more robust</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/d32d0535f7274a679c47d3354411476a86f5971a"><code>d32d053</code></a> fix SLF4J-535</li>
<li><a href="https://github.com/qos-ch/slf4j/commit/2b657bf5dc575f32791648fd95260e33aa07687c"><code>2b657bf</code></a> start work on 1.7.33-SNAPSHOT</li>
<li>See full diff in <a href="https://github.com/qos-ch/slf4j/compare/v_1.7.33...v_1.7.35">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: 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/1800#" 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"]} -->

Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 675ef46bf..600ae4aac 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -99,6 +99,7 @@ aliases:
   - matzew
   eventing-rabbitmq-approvers:
   - benmoss
+  - chunyilyu
   - gabo1208
   - gvmw
   - ikvmw
@@ -136,6 +137,7 @@ aliases:
   - matejvasek
   - nainaz
   - omerbensaadon
+  - salaboy
   - zroubalik
   kn-plugin-migration-approvers:
   - maximilien
@@ -168,6 +170,7 @@ aliases:
   - nicolaferraro
   - rhuss
   knative-admin:
+  - carlisia
   - csantanapr
   - dprotaso
   - evankanderson
@@ -178,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
@@ -228,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/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index c3a6be022..e505ac58d 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -18,15 +18,13 @@ package channel
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"strconv"
 
 	"github.com/Shopify/sarama"
+
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -39,7 +37,6 @@ import (
 	"knative.dev/pkg/system"
 
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
-	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
 
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/common/constants"
@@ -80,8 +77,7 @@ type Reconciler struct {
 	// reconciliation loop.
 	InitOffsetsFunc kafka.InitOffsetsFunc
 
-	ConfigMapLister    corelisters.ConfigMapLister
-	SubscriptionLister messaginglisters.SubscriptionLister
+	ConfigMapLister corelisters.ConfigMapLister
 
 	Prober prober.Prober
 
@@ -480,21 +476,11 @@ func (r *Reconciler) reconcileInitialOffset(ctx context.Context, channel *messag
 }
 
 func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (*contract.Egress, error) {
-	subscriptionName, err := r.getSubscriptionName(channel, subscriber)
-	if err != nil {
-		return nil, fmt.Errorf("failed to extract subscription name for subscriber %v: %w", subscriber, err)
-	}
-
 	egress := &contract.Egress{
 		Destination:   subscriber.SubscriberURI.String(),
 		ConsumerGroup: consumerGroup(channel, subscriber),
 		DeliveryOrder: DefaultDeliveryOrder,
 		Uid:           string(subscriber.UID),
-		Reference: &contract.Reference{
-			Uuid:      string(subscriber.UID),
-			Namespace: channel.GetNamespace(),
-			Name:      subscriptionName,
-		},
 		ReplyStrategy: &contract.Egress_DiscardReply{},
 	}
 
@@ -595,21 +581,6 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 	return resource, nil
 }
 
-func (r *Reconciler) getSubscriptionName(channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (string, error) {
-	subscriptions, err := r.SubscriptionLister.Subscriptions(channel.GetNamespace()).List(labels.Everything())
-	if err != nil {
-		return "", fmt.Errorf("failed to list subscriptions in namespace %s: %w", channel.GetNamespace(), err)
-	}
-
-	for _, s := range subscriptions {
-		if s.UID == subscriber.UID {
-			return s.Name, nil
-		}
-	}
-
-	return "", errors.New(string(metav1.StatusReasonNotFound))
-}
-
 // consumerGroup returns a consumerGroup name for the given channel and subscription
 func consumerGroup(channel *messagingv1beta1.KafkaChannel, sub *v1.SubscriberSpec) string {
 	return fmt.Sprintf("kafka.%s.%s.%s", channel.Namespace, channel.Name, string(sub.UID))
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index db8296757..8ccfa800b 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -25,8 +25,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
 
-	subscriptioninformer "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
-
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	kafkachannelinformer "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
 	kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
@@ -67,7 +65,6 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 			DispatcherLabel:             base.ChannelDispatcherLabel,
 			ReceiverLabel:               base.ChannelReceiverLabel,
 		},
-		SubscriptionLister:         subscriptioninformer.Get(ctx).Lister(),
 		NewKafkaClient:             sarama.NewClient,
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		InitOffsetsFunc:            offset.InitOffsets,
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/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 1aaabe7c3..e732754de 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -154,7 +154,7 @@ Lists of 158 third-party dependencies.
      (Apache-2.0) Scala Compiler (org.scala-lang:scala-reflect:2.12.13 - https://www.scala-lang.org/)
      (Apache-2.0) scala-collection-compat (org.scala-lang.modules:scala-collection-compat_2.12:2.3.0 - http://www.scala-lang.org/)
      (Apache-2.0) scala-java8-compat (org.scala-lang.modules:scala-java8-compat_2.12:0.9.1 - http://www.scala-lang.org/)
-     (MIT License) SLF4J API Module (org.slf4j:slf4j-api:1.7.33 - http://www.slf4j.org)
-     (MIT License) SLF4J NOP Binding (org.slf4j:slf4j-nop:1.7.33 - http://www.slf4j.org)
+     (MIT License) SLF4J API Module (org.slf4j:slf4j-api:1.7.35 - http://www.slf4j.org)
+     (MIT License) SLF4J NOP Binding (org.slf4j:slf4j-nop:1.7.35 - http://www.slf4j.org)
      (Apache-2.0) snappy-java (org.xerial.snappy:snappy-java:1.1.8.1 - https://github.com/xerial/snappy-java)
      (Apache License, Version 2.0) SnakeYAML (org.yaml:snakeyaml:1.28 - http://www.snakeyaml.org)
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
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index eaddaf405..910aa857a 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -47,7 +47,7 @@
     <opentelemetry.version>1.10.1</opentelemetry.version>
     <jackson.version>2.13.1</jackson.version>
     <protobuf.version>3.19.3</protobuf.version>
-    <slf4j.version>1.7.33</slf4j.version>
+    <slf4j.version>1.7.35</slf4j.version>
     <ch.qos.logback.version>1.2.10</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>7.0.1</net.logstash.logback.encoder.version>
     <assertj.version>3.22.0</assertj.version>
diff --git a/go.mod b/go.mod
index 548f5352a..378cb5b92 100644
--- a/go.mod
+++ b/go.mod
@@ -23,8 +23,8 @@ require (
 	k8s.io/apiserver v0.22.5
 	k8s.io/client-go v0.22.5
 	k8s.io/utils v0.0.0-20211208161948-7d6a63dca704
-	knative.dev/eventing v0.28.1-0.20220124132429-9e0e90c9d603
-	knative.dev/eventing-kafka v0.28.1-0.20220121065504-fe10ecf57e92
+	knative.dev/eventing v0.29.0
+	knative.dev/eventing-kafka v0.28.1-0.20220126011444-734561d578b9
 	knative.dev/hack v0.0.0-20220118141833-9b2ed8471e30
 	knative.dev/pkg v0.0.0-20220118160532-77555ea48cd4
 	knative.dev/reconciler-test v0.0.0-20220118183433-c8bfbe66bada
diff --git a/go.sum b/go.sum
index a092cd9c4..2860ff725 100644
--- a/go.sum
+++ b/go.sum
@@ -1345,11 +1345,10 @@ k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20211208161948-7d6a63dca704 h1:ZKMMxTvduyf5WUtREOqg5LiXaN1KO/+0oOQPRFrClpo=
 k8s.io/utils v0.0.0-20211208161948-7d6a63dca704/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20220118185434-ad4ce9e9d455/go.mod h1:xiTC5Qt9GdPuhRknr0Oq744SsUqbkDx0oe04L9a+qYY=
-knative.dev/eventing v0.28.1-0.20220119171353-ad895de5fdd8/go.mod h1:u5T5NZTDUsLR7yJwp5MDnBnDX5MhywD3yK3Rq+7gTtI=
-knative.dev/eventing v0.28.1-0.20220124132429-9e0e90c9d603 h1:y1ms8nohoiqUD/Nmt0ARj4ZH79jrSjjIJ16tMstH+Eo=
-knative.dev/eventing v0.28.1-0.20220124132429-9e0e90c9d603/go.mod h1:u5T5NZTDUsLR7yJwp5MDnBnDX5MhywD3yK3Rq+7gTtI=
-knative.dev/eventing-kafka v0.28.1-0.20220121065504-fe10ecf57e92 h1:Saz5jV6m4MBhFJ5uc0fNnafp70zR/Uh1iFHaCU+87RU=
-knative.dev/eventing-kafka v0.28.1-0.20220121065504-fe10ecf57e92/go.mod h1:UVCY5kWcBfyp4N6pJ84dVR0mz8pd14jMTBP4GQDjOPY=
+knative.dev/eventing v0.29.0 h1:+AhcYEuuz37x1op71bknZ/lFgsvuZJl0AKFOOFiH03s=
+knative.dev/eventing v0.29.0/go.mod h1:u5T5NZTDUsLR7yJwp5MDnBnDX5MhywD3yK3Rq+7gTtI=
+knative.dev/eventing-kafka v0.28.1-0.20220126011444-734561d578b9 h1:UCtzXoRZvZf4dXDZ7oKjXny2FLCDqIIZnYQdtu2A3/o=
+knative.dev/eventing-kafka v0.28.1-0.20220126011444-734561d578b9/go.mod h1:Y9+hHdaYCmD46PPOU2pAfWCUXJyIfgJOnkWZwsBAEoU=
 knative.dev/hack v0.0.0-20220118141833-9b2ed8471e30 h1:UkNpCWCMM5C4AeQ8aTrPTuR/6OeARiqk+LEQ6tuMP7c=
 knative.dev/hack v0.0.0-20220118141833-9b2ed8471e30/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20220118141833-9b2ed8471e30/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
deleted file mode 100644
index eb18a1fac..000000000
--- a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
-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"
-
-	fake "knative.dev/eventing/pkg/client/injection/informers/factory/fake"
-	subscription "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
-	controller "knative.dev/pkg/controller"
-	injection "knative.dev/pkg/injection"
-)
-
-var Get = subscription.Get
-
-func init() {
-	injection.Fake.RegisterInformer(withInformer)
-}
-
-func withInformer(ctx context.Context) (context.Context, controller.Informer) {
-	f := fake.Get(ctx)
-	inf := f.Messaging().V1().Subscriptions()
-	return context.WithValue(ctx, subscription.Key{}, inf), inf.Informer()
-}
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
deleted file mode 100644
index 37862d8a5..000000000
--- a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
-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 subscription
-
-import (
-	context "context"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	labels "k8s.io/apimachinery/pkg/labels"
-	cache "k8s.io/client-go/tools/cache"
-	apismessagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
-	versioned "knative.dev/eventing/pkg/client/clientset/versioned"
-	v1 "knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1"
-	client "knative.dev/eventing/pkg/client/injection/client"
-	factory "knative.dev/eventing/pkg/client/injection/informers/factory"
-	messagingv1 "knative.dev/eventing/pkg/client/listers/messaging/v1"
-	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.Messaging().V1().Subscriptions()
-	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.SubscriptionInformer {
-	untyped := ctx.Value(Key{})
-	if untyped == nil {
-		logging.FromContext(ctx).Panic(
-			"Unable to fetch knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1.SubscriptionInformer from context.")
-	}
-	return untyped.(v1.SubscriptionInformer)
-}
-
-type wrapper struct {
-	client versioned.Interface
-
-	namespace string
-
-	resourceVersion string
-}
-
-var _ v1.SubscriptionInformer = (*wrapper)(nil)
-var _ messagingv1.SubscriptionLister = (*wrapper)(nil)
-
-func (w *wrapper) Informer() cache.SharedIndexInformer {
-	return cache.NewSharedIndexInformer(nil, &apismessagingv1.Subscription{}, 0, nil)
-}
-
-func (w *wrapper) Lister() messagingv1.SubscriptionLister {
-	return w
-}
-
-func (w *wrapper) Subscriptions(namespace string) messagingv1.SubscriptionNamespaceLister {
-	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 []*apismessagingv1.Subscription, err error) {
-	lo, err := w.client.MessagingV1().Subscriptions(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) (*apismessagingv1.Subscription, error) {
-	return w.client.MessagingV1().Subscriptions(w.namespace).Get(context.TODO(), name, metav1.GetOptions{
-		ResourceVersion: w.resourceVersion,
-	})
-}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 5ba7d955c..694d7d209 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1124,7 +1124,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.28.1-0.20220124132429-9e0e90c9d603
+# knative.dev/eventing v0.29.0
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1179,8 +1179,6 @@ knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger
 knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake
 knative.dev/eventing/pkg/client/injection/informers/factory
 knative.dev/eventing/pkg/client/injection/informers/factory/fake
-knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription
-knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger
 knative.dev/eventing/pkg/client/listers/eventing/v1
@@ -1274,7 +1272,7 @@ knative.dev/eventing/test/upgrade/prober/wathola/fetcher
 knative.dev/eventing/test/upgrade/prober/wathola/forwarder
 knative.dev/eventing/test/upgrade/prober/wathola/receiver
 knative.dev/eventing/test/upgrade/prober/wathola/sender
-# knative.dev/eventing-kafka v0.28.1-0.20220121065504-fe10ecf57e92
+# knative.dev/eventing-kafka v0.28.1-0.20220126011444-734561d578b9
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1


Test patch:
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index ffd577dbc..64f9e2831 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -22,7 +22,6 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
-	eventingrekttesting "knative.dev/eventing/pkg/reconciler/testing/v1"
 	"knative.dev/pkg/network"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
@@ -351,9 +350,6 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -376,11 +372,6 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}},
 						},
 					},
@@ -434,9 +425,6 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -459,11 +447,6 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}},
 						},
 					},
@@ -517,9 +500,6 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -542,11 +522,6 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}},
 						},
 					},
@@ -603,12 +578,6 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
-				eventingrekttesting.NewSubscription(Subscription2Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription2UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -631,22 +600,12 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}, {
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription2UUID,
 								Destination:   "http://" + Subscription2URI,
 								Uid:           Subscription2UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_DiscardReply{},
-								Reference: &contract.Reference{
-									Name:      Subscription2Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription2UUID,
-								},
 							}},
 						},
 					},
@@ -974,9 +933,6 @@ func TestReconcileKind(t *testing.T) {
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
 				NewSSLSecret(system.Namespace(), "secret-1"),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -1007,11 +963,6 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}},
 						},
 					},
@@ -1266,9 +1217,8 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 				DispatcherLabel:             base.ChannelDispatcherLabel,
 				ReceiverLabel:               base.ChannelReceiverLabel,
 			},
-			Env:                env,
-			ConfigMapLister:    listers.GetConfigMapLister(),
-			SubscriptionLister: listers.GetSubscriptionLister(),
+			Env:             env,
+			ConfigMapLister: listers.GetConfigMapLister(),
 			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 				return 1, nil
 			},
diff --git a/control-plane/pkg/reconciler/channel/controller_test.go b/control-plane/pkg/reconciler/channel/controller_test.go
index f3a49c227..cc3ca16ce 100644
--- a/control-plane/pkg/reconciler/channel/controller_test.go
+++ b/control-plane/pkg/reconciler/channel/controller_test.go
@@ -24,7 +24,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake"
-	_ "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/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"
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)
+		})
+	}
+
+}