Repository: knative-sandbox/eventing-kafka-broker
Pull request #1374 (2021-10-27 07:24:08)

Kafka Channel Subscription reconciliation - control plane
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Reconcile subscribers in the control plane
- Init offsets and mark subscriptions ready
- Data plane changes (the check to see if the offsets are initialized) are to be done in a separate PR. 

<!--
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/1374#" title="Author self-approved">aliok</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **evankanderson** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @evankanderson` 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":["evankanderson"]} -->
knative-prow-robot: @aliok: 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-unit-tests | 6975dd8b131228e5e038ea469a5fc7c5b313cf36 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1455147058418683904) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<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 -->
aliok: /test pull-knative-sandbox-eventing-kafka-broker-unit-tests

Maven crap
knative-prow-robot: @aliok: 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-unit-tests | f873ade77634f25cf3dddec8b179f0c7ae224896 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1455453531635978240) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<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: @aliok: 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-unit-tests | f873ade77634f25cf3dddec8b179f0c7ae224896 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1455453531635978240) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | f873ade77634f25cf3dddec8b179f0c7ae224896 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1455453531694698496) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<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: @aliok: 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-go-coverage | f873ade77634f25cf3dddec8b179f0c7ae224896 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1455453531694698496) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 39c353db1c9d264e3f869ba04fb166b3a75ccb7f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1455454225164144640) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<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: @aliok: 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-go-coverage | f873ade77634f25cf3dddec8b179f0c7ae224896 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1455453531694698496) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 39c353db1c9d264e3f869ba04fb166b3a75ccb7f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1455454225164144640) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 39c353db1c9d264e3f869ba04fb166b3a75ccb7f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-build-tests/1455454225143173120) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<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: @aliok: 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-unit-tests | 39c353db1c9d264e3f869ba04fb166b3a75ccb7f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1455454225164144640) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 39c353db1c9d264e3f869ba04fb166b3a75ccb7f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-build-tests/1455454225143173120) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 39c353db1c9d264e3f869ba04fb166b3a75ccb7f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1455454225227059200) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<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: @aliok: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

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

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

<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 -->
aliok: /test pull-knative-sandbox-eventing-kafka-broker-integration-tests
google-cla[bot]: All (the pull request submitter and all commit authors) CLAs are signed, **but** one or more commits were authored or co-authored by someone other than the pull request submitter.

We need to confirm that all authors are ok with their commits being contributed to this project.  Please have them confirm that by leaving a comment that contains only `@googlebot I consent.` in this pull request.

*Note to project maintainer:* There may be cases where the author cannot leave a comment, or the comment is not properly detected as consent.  In those cases, you can manually confirm consent of the commit author(s), and set the `cla` label to `yes` (if enabled on your project).

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

<!-- need_author_consent -->
aliok: /close
knative-prow-robot: @aliok: Closed this PR.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/1374#issuecomment-957341988):

>/close


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>
aliok: /open
aliok: /reopen
knative-prow-robot: @aliok: Reopened this PR.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/1374#issuecomment-957342634):

>/reopen


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>
pierDipi: @googlebot I consent.
knative-prow-robot: @aliok: 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-integration-tests | 8aaa9ebf4f1e5d9349f77aadc9531f48d318d982 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1455458165381402624) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 7cd35741013eddc84237e05579f1be53a4db44b5 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1374/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1455493174997815296) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1374#" title="Author self-approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1374#pullrequestreview-795243891" title="Approved">pierDipi</a>*

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

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

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)~~ [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":[]} -->

Patch:
diff --git a/control-plane/pkg/core/config/egress.go b/control-plane/pkg/core/config/egress.go
index e18425e05..56a186875 100644
--- a/control-plane/pkg/core/config/egress.go
+++ b/control-plane/pkg/core/config/egress.go
@@ -47,11 +47,11 @@ const (
 )
 
 // AddOrUpdateEgressConfig adds or updates the given egress to the given contract at the specified indexes.
-func AddOrUpdateEgressConfig(ct *contract.Contract, brokerIndex int, egress *contract.Egress, egressIndex int) int {
+func AddOrUpdateEgressConfig(ct *contract.Contract, resourceIndex int, egress *contract.Egress, egressIndex int) int {
 
 	if egressIndex != NoEgress {
-		prev := ct.Resources[brokerIndex].Egresses[egressIndex]
-		ct.Resources[brokerIndex].Egresses[egressIndex] = egress
+		prev := ct.Resources[resourceIndex].Egresses[egressIndex]
+		ct.Resources[resourceIndex].Egresses[egressIndex] = egress
 
 		if proto.Equal(prev, egress) {
 			return EgressUnchanged
@@ -59,14 +59,32 @@ func AddOrUpdateEgressConfig(ct *contract.Contract, brokerIndex int, egress *con
 		return EgressChanged
 	}
 
-	ct.Resources[brokerIndex].Egresses = append(
-		ct.Resources[brokerIndex].Egresses,
+	ct.Resources[resourceIndex].Egresses = append(
+		ct.Resources[resourceIndex].Egresses,
 		egress,
 	)
 
 	return EgressChanged
 }
 
+// AddOrUpdateEgressConfigForResource adds or updates the given egress to the given contract at the specified indexes.
+func AddOrUpdateEgressConfigForResource(resource *contract.Resource, egress *contract.Egress, egressIndex int) int {
+
+	if egressIndex != NoEgress {
+		prev := resource.Egresses[egressIndex]
+		resource.Egresses[egressIndex] = egress
+
+		if proto.Equal(prev, egress) {
+			return EgressUnchanged
+		}
+		return EgressChanged
+	}
+
+	resource.Egresses = append(resource.Egresses, egress)
+
+	return EgressChanged
+}
+
 // KeyTypeFromString returns the contract.KeyType associated to a given string.
 func KeyTypeFromString(s string) contract.KeyType {
 	switch s {
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 29503c905..abe3cc072 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -127,7 +127,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 
 	topicName := kafka.Topic(TopicPrefix, broker)
 
-	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(securityOption)
+	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin config: %w", err))
 	}
@@ -299,7 +299,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 	// get security option for Sarama with secret info in it
 	securityOption := security.NewSaramaSecurityOptionFromSecret(secret)
 
-	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(securityOption)
+	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 	if err != nil {
 		// even in error case, we return `normal`, since we are fine with leaving the
 		// topic undeleted e.g. when we lose connection
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 1799a49f0..e420c24ea 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -23,25 +23,28 @@ import (
 	"strings"
 
 	"github.com/Shopify/sarama"
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/util/retry"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
 	"knative.dev/eventing-kafka/pkg/common/constants"
+	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
 	commonsarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
 	"knative.dev/pkg/system"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
-
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
@@ -62,6 +65,10 @@ type Reconciler struct {
 	// mock the function used during the reconciliation loop.
 	NewKafkaClusterAdmin kafka.NewClusterAdminFunc
 
+	// NewKafkaClient creates new sarama Client. It's convenient to add this as Reconciler field so that we can
+	// mock the function used during the reconciliation loop.
+	NewKafkaClient kafka.NewClientFunc
+
 	ConfigMapLister corelisters.ConfigMapLister
 
 	Configs *config.Env
@@ -135,17 +142,31 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 
 	topicName := topic(TopicPrefix, channel)
 
-	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(saramaSecurityOption)
+	kafkaClusterAdminSaramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin sarama config: %w", err))
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(topicConfig.BootstrapServers, saramaConfig)
+	// Manually commit the offsets in KafkaChannel controller.
+	// That's because we want to make sure we initialize the offsets within the controller
+	// before dispatcher actually starts consuming messages.
+	kafkaClientSaramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption, kafka.DisableOffsetAutoCommitConfigOption)
+	if err != nil {
+		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin sarama config: %w", err))
+	}
+
+	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(topicConfig.BootstrapServers, kafkaClusterAdminSaramaConfig)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("cannot obtain Kafka cluster admin, %w", err))
 	}
 	defer kafkaClusterAdmin.Close()
 
+	kafkaClient, err := r.NewKafkaClient(topicConfig.BootstrapServers, kafkaClientSaramaConfig)
+	if err != nil {
+		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting sarama config: %w", err))
+	}
+	defer kafkaClient.Close()
+
 	// create the topic
 	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdmin, logger, topicName, topicConfig)
 	if err != nil {
@@ -154,16 +175,16 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 	logger.Debug("Topic created", zap.Any("topic", topic))
 	statusConditionManager.TopicReady(topic)
 
-	// Get contract config map to write into it.
+	// Get data plane config map.
 	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
 	if err != nil {
-		return statusConditionManager.FailedToGetConfigMap(err)
+		return statusConditionManager.FailedToGetConfig(err)
 	}
 	logger.Debug("Got contract config map")
 
-	// Get contract data
+	// Get data plane config data.
 	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
-	if err != nil && ct == nil {
+	if err != nil || ct == nil {
 		return statusConditionManager.FailedToGetDataFromConfigMap(err)
 	}
 	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
@@ -175,7 +196,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 	}
 	coreconfig.SetDeadLetterSinkURIFromEgressConfig(&channel.Status.DeliveryStatus, channelResource.EgressConfig)
 
-	// Update contract data with the new contract configuration
+	// we still update the contract configMap even though there's an error.
+	// however, we record this error to make the reconciler try again.
+	subscriptionError := r.reconcileSubscribers(ctx, kafkaClient, kafkaClusterAdmin, channel, channelResource)
+
+	// Update contract data with the new contract configuration (add/update channel resource)
 	channelIndex := coreconfig.FindResource(ct, channel.UID)
 	changed := coreconfig.AddOrUpdateResourceConfig(ct, channelResource, channelIndex, logger)
 	logger.Debug("Change detector", zap.Int("changed", changed))
@@ -229,6 +254,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 		logger.Debug("Updated dispatcher pod annotation")
 	}
 
+	if subscriptionError != nil {
+		logger.Error("Error reconciling subscriptions. Going to try again.", zap.Error(subscriptionError))
+		return subscriptionError
+	}
+
 	return statusConditionManager.Reconciled()
 }
 
@@ -327,7 +357,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 	// get security option for Sarama with secret info in it
 	saramaSecurityOption := security.NewSaramaSecurityOptionFromSecret(secret)
 
-	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(saramaSecurityOption)
+	saramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption)
 	if err != nil {
 		// even in error case, we return `normal`, since we are fine with leaving the
 		// topic undeleted e.g. when we lose connection
@@ -352,6 +382,91 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 	return nil
 }
 
+func (r *Reconciler) reconcileSubscribers(ctx context.Context, kafkaClient sarama.Client, kafkaClusterAdmin sarama.ClusterAdmin, channel *messagingv1beta1.KafkaChannel, channelContractResource *contract.Resource) error {
+	logger := kafkalogging.CreateReconcileMethodLogger(ctx, channel)
+
+	channel.Status.Subscribers = make([]v1.SubscriberStatus, 0)
+	var globalErr error
+	for i := range channel.Spec.Subscribers {
+		s := &channel.Spec.Subscribers[i]
+		logger = logger.With(zap.Any("subscription", s))
+		err := r.reconcileSubscriber(ctx, kafkaClient, kafkaClusterAdmin, channel, s, channelContractResource)
+		if err != nil {
+			logger.Error("error reconciling subscription. marking subscription as not ready", zap.Error(err))
+			channel.Status.Subscribers = append(channel.Status.Subscribers, v1.SubscriberStatus{
+				UID:                s.UID,
+				ObservedGeneration: s.Generation,
+				Ready:              corev1.ConditionFalse,
+				Message:            fmt.Sprint("Subscription not ready", err),
+			})
+			globalErr = multierr.Append(globalErr, err)
+		} else {
+			logger.Debug("marking subscription as ready")
+			channel.Status.Subscribers = append(channel.Status.Subscribers, v1.SubscriberStatus{
+				UID:                s.UID,
+				ObservedGeneration: s.Generation,
+				Ready:              corev1.ConditionTrue,
+			})
+		}
+	}
+	return globalErr
+}
+
+func (r *Reconciler) reconcileSubscriber(ctx context.Context, kafkaClient sarama.Client, kafkaClusterAdmin sarama.ClusterAdmin, channel *messagingv1beta1.KafkaChannel, subscriberSpec *v1.SubscriberSpec, channelContractResource *contract.Resource) error {
+	logger := kafkalogging.CreateReconcileMethodLogger(ctx, channel)
+
+	logger.Debug("Reconciling initial offset for subscription", zap.Any("subscription", subscriberSpec), zap.Any("channel", channel))
+	err := r.reconcileInitialOffset(ctx, channel, subscriberSpec, kafkaClient, kafkaClusterAdmin)
+	if err != nil {
+		return fmt.Errorf("initial offset cannot be committed: %v", err)
+	}
+	logger.Debug("Reconciled initial offset for subscription. ", zap.Any("subscription", subscriberSpec))
+
+	subscriberIndex := coreconfig.FindEgress(channelContractResource.Egresses, subscriberSpec.UID)
+	subscriberConfig, err := r.getSubscriberConfig(ctx, channel, subscriberSpec)
+	if err != nil {
+		return fmt.Errorf("failed to resolve subscriber config: %v", zap.Error(err))
+	}
+
+	coreconfig.AddOrUpdateEgressConfigForResource(channelContractResource, subscriberConfig, subscriberIndex)
+	return nil
+}
+
+func (r *Reconciler) reconcileInitialOffset(ctx context.Context, channel *messagingv1beta1.KafkaChannel, sub *v1.SubscriberSpec, kafkaClient sarama.Client, kafkaClusterAdmin sarama.ClusterAdmin) error {
+	subscriptionStatus := findSubscriptionStatus(channel, sub.UID)
+	if subscriptionStatus != nil && subscriptionStatus.Ready == corev1.ConditionTrue {
+		// subscription is ready, the offsets must have been initialized already
+		return nil
+	}
+
+	topicName := topic(TopicPrefix, channel)
+	groupID := consumerGroup(channel, sub)
+	_, err := offset.InitOffsets(ctx, kafkaClient, kafkaClusterAdmin, []string{topicName}, groupID)
+	return err
+}
+
+func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (*contract.Egress, error) {
+	egress := &contract.Egress{
+		Destination:   subscriber.SubscriberURI.String(),
+		ConsumerGroup: consumerGroup(channel, subscriber),
+		Uid:           string(subscriber.UID),
+	}
+
+	subscriptionEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, channel, subscriber.Delivery, r.Configs.DefaultBackoffDelayMs)
+	if err != nil {
+		return nil, err
+	}
+	channelEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, channel, channel.Spec.Delivery, r.Configs.DefaultBackoffDelayMs)
+	if err != nil {
+		return nil, err
+	}
+
+	// Merge Channel and Subscription egress configuration prioritizing the Subscription configuration.
+	egress.EgressConfig = coreconfig.MergeEgressConfig(subscriptionEgressConfig, channelEgressConfig)
+
+	return egress, nil
+}
+
 func (r *Reconciler) channelConfigMap() (*corev1.ConfigMap, error) {
 	// TODO: do we want to support namespaced channels? they're not supported at the moment.
 
@@ -422,8 +537,22 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 	return resource, nil
 }
 
-// Topic returns a topic name given a topic prefix and a generic object.
+// topic returns a topic name given a topic prefix and a generic object.
 // This function uses a different format than the kafkatopic.Topic function
 func topic(prefix string, obj metav1.Object) string {
 	return fmt.Sprintf("%s.%s.%s", prefix, obj.GetNamespace(), obj.GetName())
 }
+
+// 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))
+}
+
+func findSubscriptionStatus(kc *messagingv1beta1.KafkaChannel, subUID types.UID) *v1.SubscriberStatus {
+	for _, subStatus := range kc.Status.Subscribers {
+		if subStatus.UID == subUID {
+			return &subStatus
+		}
+	}
+	return nil
+}
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index dd7cb44c2..ec4f1b87d 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -60,6 +60,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 			DispatcherLabel:             base.ChannelDispatcherLabel,
 			ReceiverLabel:               base.ChannelReceiverLabel,
 		},
+		NewKafkaClient:       sarama.NewClient,
 		NewKafkaClusterAdmin: sarama.NewClusterAdmin,
 		Configs:              configs,
 		ConfigMapLister:      configmapInformer.Lister(),
diff --git a/control-plane/pkg/reconciler/kafka/client.go b/control-plane/pkg/reconciler/kafka/client.go
index fdf598b29..10f3de2b5 100644
--- a/control-plane/pkg/reconciler/kafka/client.go
+++ b/control-plane/pkg/reconciler/kafka/client.go
@@ -36,15 +36,23 @@ func NoOpConfigOption(*sarama.Config) error {
 	return nil
 }
 
+func DisableOffsetAutoCommitConfigOption(config *sarama.Config) error {
+	config.Consumer.Offsets.AutoCommit.Enable = false
+	return nil
+}
+
 // NewClusterAdminFunc creates new sarama.ClusterAdmin.
 type NewClusterAdminFunc func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error)
 
-// GetClusterAdminSaramaConfig returns Kafka Admin configurations that the ConfigOptions are applied to
-func GetClusterAdminSaramaConfig(configOption ConfigOption) (*sarama.Config, error) {
+// NewClientFunc creates new sarama.Client.
+type NewClientFunc func(addrs []string, config *sarama.Config) (sarama.Client, error)
+
+// GetSaramaConfig returns Kafka Client configuration with the given options applied.
+func GetSaramaConfig(configOptions ...ConfigOption) (*sarama.Config, error) {
 	config := sarama.NewConfig()
 	config.Version = sarama.MaxVersion
 
-	err := configOption(config)
+	err := Options(config, configOptions...)
 	if err != nil {
 		return nil, err
 	}
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index bee978514..ea1ec7ef8 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -100,7 +100,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		return fmt.Errorf("failed to track secret: %w", err)
 	}
 
-	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(securityOption)
+	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 	if err != nil {
 		return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 	}
@@ -286,7 +286,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 		// get security option for Sarama with secret info in it
 		securityOption := security.NewSaramaSecurityOptionFromSecret(secret)
 
-		saramaConfig, err := kafka.GetClusterAdminSaramaConfig(securityOption)
+		saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 		if err != nil {
 			// even in error case, we return `normal`, since we are fine with leaving the
 			// topic undeleted e.g. when we lose connection
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index 125cecb34..4ed816d0a 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -99,7 +99,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *sources.KafkaSource)
 		return fmt.Errorf("failed to track secrets: %w", err)
 	}
 
-	saramaConfig, err := kafka.GetClusterAdminSaramaConfig(securityOption)
+	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 	if err != nil {
 		return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 	}
diff --git a/go.mod b/go.mod
index 6fe198b8e..f7b62456f 100644
--- a/go.mod
+++ b/go.mod
@@ -13,6 +13,7 @@ require (
 	github.com/rickb777/date v1.14.1
 	github.com/stretchr/testify v1.7.0
 	github.com/xdg-go/scram v1.0.2
+	go.uber.org/multierr v1.6.0
 	go.uber.org/zap v1.19.1
 	google.golang.org/protobuf v1.27.1
 	k8s.io/api v0.21.4
diff --git a/vendor/knative.dev/eventing-kafka/pkg/common/kafka/offset/offsets.go b/vendor/knative.dev/eventing-kafka/pkg/common/kafka/offset/offsets.go
new file mode 100644
index 000000000..5caab1698
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/common/kafka/offset/offsets.go
@@ -0,0 +1,139 @@
+/*
+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 offset
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/Shopify/sarama"
+	"go.uber.org/zap"
+	"knative.dev/pkg/logging"
+
+	knsarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
+)
+
+// We want to make sure that ALL consumer group offsets are set before marking
+// the resource as ready, to avoid "losing" events in case the consumer group session
+// is closed before at least one message is consumed from ALL partitions.
+// Without InitOffsets, an event sent to a partition with an uninitialized offset
+// will not be forwarded when the session is closed (or a rebalancing is in progress).
+func InitOffsets(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
+	offsetManager, err := sarama.NewOffsetManagerFromClient(consumerGroup, kafkaClient)
+	if err != nil {
+		return -1, err
+	}
+
+	totalPartitions, topicPartitions, err := retrieveAllPartitions(topics, kafkaClient)
+	if err != nil {
+		return -1, err
+	}
+
+	// Fetch topic offsets
+	topicOffsets, err := knsarama.GetOffsets(kafkaClient, topicPartitions, sarama.OffsetNewest)
+	if err != nil {
+		return -1, fmt.Errorf("failed to get the topic offsets: %w", err)
+	}
+
+	// Look for uninitialized offset (-1)
+	offsets, err := kafkaAdminClient.ListConsumerGroupOffsets(consumerGroup, topicPartitions)
+	if err != nil {
+		return -1, err
+	}
+
+	dirty := false
+	for topic, partitions := range offsets.Blocks {
+		for partitionID, block := range partitions {
+			if block.Offset == -1 { // not initialized?
+				partitionsOffsets, ok := topicOffsets[topic]
+				if !ok {
+					// topic may have been deleted. ignore.
+					continue
+				}
+
+				offset, ok := partitionsOffsets[partitionID]
+				if !ok {
+					// partition may have been deleted. ignore.
+					continue
+				}
+
+				logging.FromContext(ctx).Infow("initializing offset", zap.String("topic", topic), zap.Int32("partition", partitionID), zap.Int64("offset", offset))
+
+				pm, err := offsetManager.ManagePartition(topic, partitionID)
+				if err != nil {
+					return -1, fmt.Errorf("failed to create the partition manager for topic %s and partition %d: %w", topic, partitionID, err)
+				}
+
+				pm.MarkOffset(offset, "")
+				dirty = true
+			}
+		}
+	}
+
+	if dirty {
+		offsetManager.Commit()
+		logging.FromContext(ctx).Infow("consumer group offsets committed", zap.String("consumergroup", consumerGroup))
+	}
+
+	// At this stage the resource is considered Ready
+	return int32(totalPartitions), nil
+
+}
+
+func CheckIfAllOffsetsInitialized(kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (bool, error) {
+	_, topicPartitions, err := retrieveAllPartitions(topics, kafkaClient)
+	if err != nil {
+		return false, err
+	}
+
+	// Look for uninitialized offset (-1)
+	offsets, err := kafkaAdminClient.ListConsumerGroupOffsets(consumerGroup, topicPartitions)
+	if err != nil {
+		return false, err
+	}
+
+	for _, partitions := range offsets.Blocks {
+		for _, block := range partitions {
+			if block.Offset == -1 { // not initialized?
+				return false, nil
+			}
+		}
+	}
+
+	return true, nil
+}
+
+func retrieveAllPartitions(topics []string, kafkaClient sarama.Client) (int, map[string][]int32, error) {
+	totalPartitions := 0
+
+	// Retrieve all partitions
+	topicPartitions := make(map[string][]int32)
+	for _, topic := range topics {
+		partitions, err := kafkaClient.Partitions(topic)
+		totalPartitions += len(partitions)
+		if err != nil {
+			return -1, nil, fmt.Errorf("failed to get partitions for topic %s: %w", topic, err)
+		}
+
+		// return a copy of the partitions array in the map
+		// Sarama is caching this array and we don't want nobody to mess with it
+		clone := make([]int32, len(partitions))
+		copy(clone, partitions)
+		topicPartitions[topic] = clone
+	}
+	return totalPartitions, topicPartitions, nil
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 15777f163..2377d7cd1 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -359,6 +359,7 @@ go.uber.org/automaxprocs/internal/cgroups
 go.uber.org/automaxprocs/internal/runtime
 go.uber.org/automaxprocs/maxprocs
 # go.uber.org/multierr v1.6.0
+## explicit
 go.uber.org/multierr
 # go.uber.org/zap v1.19.1
 ## explicit
@@ -1187,6 +1188,7 @@ knative.dev/eventing-kafka/pkg/client/listers/sources/v1beta1
 knative.dev/eventing-kafka/pkg/common/client
 knative.dev/eventing-kafka/pkg/common/config
 knative.dev/eventing-kafka/pkg/common/constants
+knative.dev/eventing-kafka/pkg/common/kafka/offset
 knative.dev/eventing-kafka/pkg/common/kafka/sarama
 # knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e
 ## explicit


Test patch:
diff --git a/control-plane/pkg/core/config/egress_test.go b/control-plane/pkg/core/config/egress_test.go
index 29f8c404b..8b893db1c 100644
--- a/control-plane/pkg/core/config/egress_test.go
+++ b/control-plane/pkg/core/config/egress_test.go
@@ -221,6 +221,96 @@ func TestAddOrUpdateEgressConfig(t *testing.T) {
 	}
 }
 
+func TestAddOrUpdateEgressConfigForResource(t *testing.T) {
+	tests := []struct {
+		name        string
+		givenRes    *contract.Resource
+		egress      *contract.Egress
+		egressIndex int
+		changed     int
+		wantRes     *contract.Resource
+	}{
+		{
+			name: "Egress found - changed",
+			givenRes: &contract.Resource{
+				Egresses: []*contract.Egress{
+					{
+						Uid: "xyz",
+					},
+				},
+			},
+			egress: &contract.Egress{
+				Uid: "abc",
+			},
+			egressIndex: 0,
+			wantRes: &contract.Resource{
+				Egresses: []*contract.Egress{
+					{
+						Uid: "abc",
+					},
+				},
+			},
+		},
+		{
+			name: "Egress found - unchanged",
+			givenRes: &contract.Resource{
+				Egresses: []*contract.Egress{
+					{
+						Uid: "abc",
+					},
+				},
+			},
+			egress: &contract.Egress{
+				Uid: "abc",
+			},
+			egressIndex: 0,
+			wantRes: &contract.Resource{
+				Egresses: []*contract.Egress{
+					{
+						Uid: "abc",
+					},
+				},
+			},
+			changed: EgressUnchanged,
+		},
+		{
+			name: "Egress not found",
+			givenRes: &contract.Resource{
+				Egresses: []*contract.Egress{
+					{
+						Uid: "abc",
+					},
+				},
+			},
+			egress: &contract.Egress{
+				Uid: "abc",
+			},
+			egressIndex: NoEgress,
+			wantRes: &contract.Resource{
+				Egresses: []*contract.Egress{
+					{
+						Uid: "abc",
+					},
+					{
+						Uid: "abc",
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := AddOrUpdateEgressConfigForResource(tt.givenRes, tt.egress, tt.egressIndex); got != tt.changed {
+				t.Errorf("AddOrUpdateEgressConfig() = %v, want %v", got, tt.changed)
+			}
+
+			if diff := cmp.Diff(tt.wantRes, tt.givenRes, protocmp.Transform()); diff != "" {
+				t.Errorf("(-want, +got) %s", diff)
+			}
+		})
+	}
+}
+
 func TestKeyTypeFromString(t *testing.T) {
 	tests := []struct {
 		s    string