Repository: knative-sandbox/eventing-kafka-broker
Pull request #1398 (2021-11-02 14:38:12)

Let's use NewKafkaClusterAdminClient as the name to avoid confusion
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- as per title
-

<!--
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 **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1398#" title="Author self-approved">matzew</a>*

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1398](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (a5fa3fb) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/83d79a0e5de55a4963f7f5b1f94ec654bb4492cd?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (83d79a0) will **decrease** coverage by `6.51%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398/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/1398?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    #1398      +/-   ##
============================================
- Coverage     76.57%   70.05%   -6.52%     
============================================
  Files            99       34      -65     
  Lines          3607     1683    -1924     
  Branches        165        0     -165     
============================================
- Hits           2762     1179    -1583     
+ Misses          635      381     -254     
+ Partials        210      123      -87     
```

| 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/1398?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/reconciler/kafka/client.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS9jbGllbnQuZ28=) | `0.00% <ø> (ø)` | |
| [control-plane/pkg/reconciler/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyLmdv) | `70.29% <100.00%> (ø)` | |
| [control-plane/pkg/reconciler/broker/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvY29udHJvbGxlci5nbw==) | `76.05% <100.00%> (ø)` | |
| [control-plane/pkg/reconciler/sink/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2NvbnRyb2xsZXIuZ28=) | `72.34% <100.00%> (ø)` | |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | `67.83% <100.00%> (ø)` | |
| [control-plane/pkg/reconciler/source/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvY29udHJvbGxlci5nbw==) | `100.00% <100.00%> (ø)` | |
| [control-plane/pkg/reconciler/source/source.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2Uvc291cmNlLmdv) | `57.39% <100.00%> (ø)` | |
| [...fka/broker/core/security/PlaintextCredentials.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvUGxhaW50ZXh0Q3JlZGVudGlhbHMuamF2YQ==) | | |
| [...enting/kafka/broker/core/OrderedAsyncExecutor.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvT3JkZXJlZEFzeW5jRXhlY3V0b3IuamF2YQ==) | | |
| [...kafka/broker/dispatcher/impl/filter/AnyFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQW55RmlsdGVyLmphdmE=) | | |
| ... and [62 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398/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/1398?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/1398?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [83d79a0...a5fa3fb](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1398?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: @matzew: 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 | 8a57feda61a76fa1d45abbd7c33ccbe385964f18 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1398/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1455546538670428160) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

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

Patch:
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index abe3cc072..ac3b55333 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -65,9 +65,9 @@ type Reconciler struct {
 	bootstrapServersLock         sync.RWMutex
 	ConfigMapLister              corelisters.ConfigMapLister
 
-	// NewKafkaClusterAdmin creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
+	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
-	NewKafkaClusterAdmin kafka.NewClusterAdminFunc
+	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
 
 	Configs *Configs
 }
@@ -132,13 +132,13 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin config: %w", err))
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(topicConfig.BootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, saramaConfig)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("cannot obtain Kafka cluster admin, %w", err))
 	}
-	defer kafkaClusterAdmin.Close()
+	defer kafkaClusterAdminClient.Close()
 
-	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdmin, logger, topicName, topicConfig)
+	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdminClient, logger, topicName, topicConfig)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topic, err)
 	}
@@ -306,15 +306,15 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 		return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(topicConfig.BootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, saramaConfig)
 	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
 		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 	}
-	defer kafkaClusterAdmin.Close()
+	defer kafkaClusterAdminClient.Close()
 
-	topic, err := kafka.DeleteTopic(kafkaClusterAdmin, kafka.Topic(TopicPrefix, broker))
+	topic, err := kafka.DeleteTopic(kafkaClusterAdminClient, kafka.Topic(TopicPrefix, broker))
 	if err != nil {
 		return err
 	}
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index 2b38c2d4a..8b23ebc89 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -71,7 +71,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 			DispatcherLabel:             base.BrokerDispatcherLabel,
 			ReceiverLabel:               base.BrokerReceiverLabel,
 		},
-		NewKafkaClusterAdmin: sarama.NewClusterAdmin,
+		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		KafkaDefaultTopicDetails: sarama.TopicDetail{
 			NumPartitions:     DefaultNumPartitions,
 			ReplicationFactor: DefaultReplicationFactor,
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index e420c24ea..ab4ac8725 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -61,9 +61,9 @@ type Reconciler struct {
 
 	Resolver *resolver.URIResolver
 
-	// NewKafkaClusterAdmin creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
+	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
-	NewKafkaClusterAdmin kafka.NewClusterAdminFunc
+	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
 
 	// 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.
@@ -155,11 +155,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin sarama config: %w", err))
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(topicConfig.BootstrapServers, kafkaClusterAdminSaramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, kafkaClusterAdminSaramaConfig)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("cannot obtain Kafka cluster admin, %w", err))
 	}
-	defer kafkaClusterAdmin.Close()
+	defer kafkaClusterAdminClient.Close()
 
 	kafkaClient, err := r.NewKafkaClient(topicConfig.BootstrapServers, kafkaClientSaramaConfig)
 	if err != nil {
@@ -168,7 +168,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 	defer kafkaClient.Close()
 
 	// create the topic
-	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdmin, logger, topicName, topicConfig)
+	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdminClient, logger, topicName, topicConfig)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topic, err)
 	}
@@ -198,7 +198,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 
 	// 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)
+	subscriptionError := r.reconcileSubscribers(ctx, kafkaClient, kafkaClusterAdminClient, channel, channelResource)
 
 	// Update contract data with the new contract configuration (add/update channel resource)
 	channelIndex := coreconfig.FindResource(ct, channel.UID)
@@ -364,15 +364,15 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 		return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(topicConfig.BootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, saramaConfig)
 	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
 		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 	}
-	defer kafkaClusterAdmin.Close()
+	defer kafkaClusterAdminClient.Close()
 
-	topic, err := kafka.DeleteTopic(kafkaClusterAdmin, topic(TopicPrefix, channel))
+	topic, err := kafka.DeleteTopic(kafkaClusterAdminClient, topic(TopicPrefix, channel))
 	if err != nil {
 		return err
 	}
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index ec4f1b87d..56eb80218 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -60,10 +60,10 @@ 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(),
+		NewKafkaClient:             sarama.NewClient,
+		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
+		Configs:                    configs,
+		ConfigMapLister:            configmapInformer.Lister(),
 	}
 
 	logger := logging.FromContext(ctx)
diff --git a/control-plane/pkg/reconciler/kafka/client.go b/control-plane/pkg/reconciler/kafka/client.go
index 10f3de2b5..b129275b5 100644
--- a/control-plane/pkg/reconciler/kafka/client.go
+++ b/control-plane/pkg/reconciler/kafka/client.go
@@ -41,8 +41,8 @@ func DisableOffsetAutoCommitConfigOption(config *sarama.Config) error {
 	return nil
 }
 
-// NewClusterAdminFunc creates new sarama.ClusterAdmin.
-type NewClusterAdminFunc func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error)
+// NewClusterAdminClientFunc creates new sarama.ClusterAdmin.
+type NewClusterAdminClientFunc func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error)
 
 // NewClientFunc creates new sarama.Client.
 type NewClientFunc func(addrs []string, config *sarama.Config) (sarama.Client, error)
diff --git a/control-plane/pkg/reconciler/sink/controller.go b/control-plane/pkg/reconciler/sink/controller.go
index c47e41965..aa4784957 100644
--- a/control-plane/pkg/reconciler/sink/controller.go
+++ b/control-plane/pkg/reconciler/sink/controller.go
@@ -58,9 +58,9 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 			SystemNamespace:             configs.SystemNamespace,
 			ReceiverLabel:               base.SinkReceiverLabel,
 		},
-		ConfigMapLister:      configmapInformer.Lister(),
-		NewKafkaClusterAdmin: sarama.NewClusterAdmin,
-		Configs:              configs,
+		ConfigMapLister:            configmapInformer.Lister(),
+		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
+		Configs:                    configs,
 	}
 
 	_, err := reconciler.GetOrCreateDataPlaneConfigMap(ctx)
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index ea1ec7ef8..26e0744e0 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -48,9 +48,9 @@ type Reconciler struct {
 
 	ConfigMapLister corelisters.ConfigMapLister
 
-	// NewKafkaClusterAdmin creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
+	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
-	NewKafkaClusterAdmin kafka.NewClusterAdminFunc
+	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
 
 	Configs *config.Env
 }
@@ -105,11 +105,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(ks.Spec.BootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(ks.Spec.BootstrapServers, saramaConfig)
 	if err != nil {
 		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 	}
-	defer kafkaClusterAdmin.Close()
+	defer kafkaClusterAdminClient.Close()
 
 	if ks.Spec.NumPartitions != nil && ks.Spec.ReplicationFactor != nil {
 
@@ -117,7 +117,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 
 		topicConfig := topicConfigFromSinkSpec(&ks.Spec)
 
-		topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdmin, logger, ks.Spec.Topic, topicConfig)
+		topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdminClient, logger, ks.Spec.Topic, topicConfig)
 		if err != nil {
 			return statusConditionManager.FailedToCreateTopic(topic, err)
 		}
@@ -126,7 +126,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		// If the topic is externally managed, we need to make sure that the topic exists and it's valid.
 		ks.GetStatus().Annotations[base.TopicOwnerAnnotation] = ExternalTopicOwner
 
-		isPresentAndValid, err := kafka.AreTopicsPresentAndValid(kafkaClusterAdmin, ks.Spec.Topic)
+		isPresentAndValid, err := kafka.AreTopicsPresentAndValid(kafkaClusterAdminClient, ks.Spec.Topic)
 		if err != nil {
 			return statusConditionManager.TopicsNotPresentOrInvalidErr([]string{ks.Spec.Topic}, err)
 		}
@@ -293,15 +293,15 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 			return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 		}
 
-		kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(ks.Spec.BootstrapServers, saramaConfig)
+		kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(ks.Spec.BootstrapServers, saramaConfig)
 		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
 			return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 		}
-		defer kafkaClusterAdmin.Close()
+		defer kafkaClusterAdminClient.Close()
 
-		topic, err := kafka.DeleteTopic(kafkaClusterAdmin, ks.Spec.Topic)
+		topic, err := kafka.DeleteTopic(kafkaClusterAdminClient, ks.Spec.Topic)
 		if err != nil {
 			return err
 		}
diff --git a/control-plane/pkg/reconciler/source/controller.go b/control-plane/pkg/reconciler/source/controller.go
index 123c6000c..875e81831 100644
--- a/control-plane/pkg/reconciler/source/controller.go
+++ b/control-plane/pkg/reconciler/source/controller.go
@@ -50,8 +50,8 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 			SystemNamespace:             configs.SystemNamespace,
 			DispatcherLabel:             base.SourceDispatcherLabel,
 		},
-		Env:                  configs,
-		NewKafkaClusterAdmin: sarama.NewClusterAdmin,
+		Env:                        configs,
+		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 	}
 
 	impl := kafkasource.NewImpl(ctx, r)
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index 4ed816d0a..133c5d3b7 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -57,9 +57,9 @@ type Reconciler struct {
 
 	Resolver *resolver.URIResolver
 
-	// NewKafkaClusterAdmin creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
+	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
-	NewKafkaClusterAdmin kafka.NewClusterAdminFunc
+	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
@@ -104,13 +104,13 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *sources.KafkaSource)
 		return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 	}
 
-	kafkaClusterAdmin, err := r.NewKafkaClusterAdmin(ks.Spec.BootstrapServers, saramaConfig)
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(ks.Spec.BootstrapServers, saramaConfig)
 	if err != nil {
 		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
 	}
-	defer kafkaClusterAdmin.Close()
+	defer kafkaClusterAdminClient.Close()
 
-	isValid, err := kafka.AreTopicsPresentAndValid(kafkaClusterAdmin, ks.Spec.Topics...)
+	isValid, err := kafka.AreTopicsPresentAndValid(kafkaClusterAdminClient, ks.Spec.Topics...)
 	if err != nil {
 		return statusConditionManager.TopicsNotPresentOrInvalidErr(ks.Spec.Topics, err)
 	}
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index befcbe197..1d6290ec6 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -20,8 +20,8 @@ Lists of 153 third-party dependencies.
      (Apache License, Version 2.0) Guava: Google Core Libraries for Java (com.google.guava:guava:30.1.1-android - https://github.com/google/guava/guava)
      (The Apache Software License, Version 2.0) Guava ListenableFuture only (com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava - https://github.com/google/guava/listenablefuture)
      (The Apache Software License, Version 2.0) J2ObjC Annotations (com.google.j2objc:j2objc-annotations:1.3 - https://github.com/google/j2objc/)
-     (3-Clause BSD License) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.19.0 - https://developers.google.com/protocol-buffers/protobuf-java/)
-     (3-Clause BSD License) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.19.0 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
+     (3-Clause BSD License) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.19.1 - https://developers.google.com/protocol-buffers/protobuf-java/)
+     (3-Clause BSD License) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.19.1 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
      (Apache 2.0) OkHttp Logging Interceptor (com.squareup.okhttp3:logging-interceptor:3.12.12 - https://github.com/square/okhttp/logging-interceptor)
      (Apache 2.0) MockWebServer (com.squareup.okhttp3:mockwebserver:3.12.6 - https://github.com/square/okhttp/mockwebserver)
      (Apache 2.0) OkHttp (com.squareup.okhttp3:okhttp:3.14.9 - https://github.com/square/okhttp/okhttp)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 6c8c43358..fd8234347 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -46,7 +46,7 @@
     <micrometer.version>1.7.5</micrometer.version>
     <opentelemetry.version>1.7.0</opentelemetry.version>
     <jackson.version>2.13.0</jackson.version>
-    <protobuf.version>3.19.0</protobuf.version>
+    <protobuf.version>3.19.1</protobuf.version>
     <slf4j.version>1.7.32</slf4j.version>
     <ch.qos.logback.version>1.2.6</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>6.6</net.logstash.logback.encoder.version>
diff --git a/go.mod b/go.mod
index f7b62456f..acfcc71b2 100644
--- a/go.mod
+++ b/go.mod
@@ -22,9 +22,9 @@ require (
 	k8s.io/apiserver v0.21.4
 	k8s.io/client-go v0.21.4
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.26.1-0.20211029085753-34c08a406eca
-	knative.dev/eventing-kafka v0.26.1-0.20211029081552-c380a0634e9c
-	knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e
-	knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40
-	knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c
+	knative.dev/eventing v0.27.0
+	knative.dev/eventing-kafka v0.27.0
+	knative.dev/hack v0.0.0-20211101195839-11d193bf617b
+	knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc
+	knative.dev/reconciler-test v0.0.0-20211101214439-9839937c9b13
 )
diff --git a/go.sum b/go.sum
index 058e124f9..f3f86d054 100644
--- a/go.sum
+++ b/go.sum
@@ -24,7 +24,6 @@ cloud.google.com/go v0.87.0/go.mod h1:TpDYlFy7vuLzZMMZ+B6iRiELaY7z/gJPaqbMx6mlWc
 cloud.google.com/go v0.90.0/go.mod h1:kRX0mNRHe0e2rC6oNakvwQqzyDmg57xJ+SZU1eT2aDQ=
 cloud.google.com/go v0.93.3/go.mod h1:8utlLll2EF5XMAV15woO4lSbWQlk8rer9aLOfLh7+YI=
 cloud.google.com/go v0.94.1/go.mod h1:qAlAugsXlC+JWO+Bke5vCtc9ONxjQT3drlTTnAplMW4=
-cloud.google.com/go v0.96.0/go.mod h1:GF7l59pYBVlXQIBLx3a761cZ41F9bBH3JUlihCt2Udc=
 cloud.google.com/go v0.97.0 h1:3DXvAyifywvq64LfkKaMOmkWPS1CikIQdMe2lY9vxU8=
 cloud.google.com/go v0.97.0/go.mod h1:GF7l59pYBVlXQIBLx3a761cZ41F9bBH3JUlihCt2Udc=
 cloud.google.com/go/bigquery v1.0.1/go.mod h1:i/xbL2UlR5RvWAURpBYZTtm/cXjCha9lbfbpx4poX+o=
@@ -45,7 +44,6 @@ cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0Zeo
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
 cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
-cloud.google.com/go/storage v1.16.1/go.mod h1:LaNorbty3ehnU3rEjXSNV/NRgQA0O8Y+uh6bPe5UOk4=
 cloud.google.com/go/storage v1.18.2 h1:5NQw6tOn3eMm0oE8vTkfjau18kjL79FlMjy/CHTpmoY=
 cloud.google.com/go/storage v1.18.2/go.mod h1:AiIj7BWXyhO5gGVmYJ+S8tbkCx3yb0IMjua8Aw4naVM=
 contrib.go.opencensus.io/exporter/ocagent v0.7.1-0.20200907061046-05415f1de66d h1:LblfooH1lKOpp1hIhukktmSAxFkqMPFk9KR6iZ0MJNI=
@@ -613,7 +611,6 @@ github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y8
 github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo=
 github.com/prometheus/common v0.26.0/go.mod h1:M7rCNAaPfAosfx8veZJCuw84e35h3Cfd9VFqTh1DIvc=
 github.com/prometheus/common v0.28.0/go.mod h1:vu+V0TpY+O6vW9J44gczi3Ap/oXXR10b+M/gUGO4Hls=
-github.com/prometheus/common v0.30.0/go.mod h1:vu+V0TpY+O6vW9J44gczi3Ap/oXXR10b+M/gUGO4Hls=
 github.com/prometheus/common v0.31.1 h1:d18hG4PkHnNAKNMOmFuXFaiY8Us0nird/2m60uS1AMs=
 github.com/prometheus/common v0.31.1/go.mod h1:vu+V0TpY+O6vW9J44gczi3Ap/oXXR10b+M/gUGO4Hls=
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
@@ -740,7 +737,6 @@ go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/automaxprocs v1.3.0/go.mod h1:9CWT6lKIep8U41DDaPiH6eFscnTyjfTANNQNx6LrIcA=
 go.uber.org/automaxprocs v1.4.0 h1:CpDZl6aOlLhReez+8S3eEotD7Jx0Os++lemPlMULQP0=
 go.uber.org/automaxprocs v1.4.0/go.mod h1:/mTEdr7LvHhs0v7mjdxDreTz1OG5zdZGqgOnhWiR/+Q=
-go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A=
 go.uber.org/goleak v1.1.11-0.20210813005559-691160354723 h1:sHOAIxRGBp443oHZIPB+HsUGaksVCXVQENPxwTfQdH4=
 go.uber.org/goleak v1.1.11-0.20210813005559-691160354723/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
@@ -750,7 +746,6 @@ go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9i
 go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA=
 go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 go.uber.org/zap v1.15.0/go.mod h1:Mb2vm2krFEG5DV0W9qcHBYFtp/Wku1cvYaqPsS/WYfc=
-go.uber.org/zap v1.19.0/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI=
 go.uber.org/zap v1.19.1 h1:ue41HOKd1vGURxrmeKIgELGb3jPW9DMUDGtsinblHwI=
 go.uber.org/zap v1.19.1/go.mod h1:j3DNczoxDZroyBnOT1L/Q79cfUMGZxlv/9dzN7SM1rI=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
@@ -865,8 +860,6 @@ golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420/go.mod h1:9nx3DQGgdP8bBQD5qx
 golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20210614182718-04defd469f4e/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20210825183410-e898025ed96a/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20210928044308-7d9f5e0b762b/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f h1:OfiFi4JbukWwe3lzw+xunroH1mnC1e2Gy5cxNJApiSY=
 golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -1018,7 +1011,6 @@ golang.org/x/tools v0.0.0-20190911174233-4f2ddba30aff/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
-golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191113191852-77e3bb0ad9e7/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191115202509-3a792d9c32b2/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
@@ -1158,7 +1150,6 @@ google.golang.org/genproto v0.0.0-20210303154014-9728d6b83eeb/go.mod h1:FWY/as6D
 google.golang.org/genproto v0.0.0-20210310155132-4ce2db91004e/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210319143718-93e7006c17a6/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210402141018-6c239bbf2bb1/go.mod h1:9lPAdzaEmUacj36I+k7YKbEc5CXzPIeORRgDAUOu28A=
-google.golang.org/genproto v0.0.0-20210416161957-9910b6c460de/go.mod h1:P3QM42oQyzQSnHPnZ/vqoCdDmzH28fzWByN9asMeM8A=
 google.golang.org/genproto v0.0.0-20210513213006-bf773b8c8384/go.mod h1:P3QM42oQyzQSnHPnZ/vqoCdDmzH28fzWByN9asMeM8A=
 google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c/go.mod h1:UODoCrxHCcBojKKwX1terBiRUaqAsFqJiF615XL43r0=
 google.golang.org/genproto v0.0.0-20210604141403-392c879c8b08/go.mod h1:UODoCrxHCcBojKKwX1terBiRUaqAsFqJiF615XL43r0=
@@ -1170,7 +1161,6 @@ google.golang.org/genproto v0.0.0-20210728212813-7823e685a01f/go.mod h1:ob2IJxKr
 google.golang.org/genproto v0.0.0-20210805201207-89edb61ffb67/go.mod h1:ob2IJxKrgPT52GcgX759i1sleT07tiKowYBGbczaW48=
 google.golang.org/genproto v0.0.0-20210813162853-db860fec028c/go.mod h1:cFeNkxwySK631ADgubI+/XFU/xp8FD5KIVV4rj8UC5w=
 google.golang.org/genproto v0.0.0-20210821163610-241b8fcbd6c8/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
-google.golang.org/genproto v0.0.0-20210825212027-de86158e7fda/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
 google.golang.org/genproto v0.0.0-20210828152312-66f60bf46e71/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
 google.golang.org/genproto v0.0.0-20210831024726-fe130286e0e2/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
 google.golang.org/genproto v0.0.0-20210903162649-d08c68adba83/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
@@ -1288,7 +1278,6 @@ k8s.io/component-base v0.21.4/go.mod h1:ZKG0eHVX+tUDcaoIGpU3Vtk4TIjMddN9uhEWDmW6
 k8s.io/gengo v0.0.0-20200413195148-3a45101e95ac/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/gengo v0.0.0-20201203183100-97869a43a9d9/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
 k8s.io/gengo v0.0.0-20201214224949-b6c5ce23f027/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
-k8s.io/gengo v0.0.0-20210203185629-de9496dff47b/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
 k8s.io/gengo v0.0.0-20210915205010-39e73c8a59cd h1:WEhFhsVrgII9WzLqf1VTUUtwPHm+vxbRq8Ib6y9Jt4g=
 k8s.io/gengo v0.0.0-20210915205010-39e73c8a59cd/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
 k8s.io/klog v0.2.0/go.mod h1:Gq+BEi5rUBO/HRz0bTSXDUcqjScdoY3a9IHpCEIOOfk=
@@ -1304,29 +1293,18 @@ k8s.io/kube-openapi v0.0.0-20210305001622-591a79e4bda7/go.mod h1:wXW5VT87nVfh/iL
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKUQWqiI/YuiBNMiQ=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/control-protocol v0.0.0-20211019131736-762838ee9058/go.mod h1:9mkxOcLsgMaA+LTojeXLpVoN1PW5jogk53pBXzrOQU4=
-knative.dev/eventing v0.26.1-0.20211027202426-729e39fd4218/go.mod h1:27s/uaaUrAUKO/Wy7kR4R6mbj5zugQWCyEkod4dVw0I=
-knative.dev/eventing v0.26.1-0.20211029085753-34c08a406eca h1:wELJavauMmbGQQ257fbAkIcZKl+AzH9fwZtD87556iA=
-knative.dev/eventing v0.26.1-0.20211029085753-34c08a406eca/go.mod h1:27s/uaaUrAUKO/Wy7kR4R6mbj5zugQWCyEkod4dVw0I=
-knative.dev/eventing-kafka v0.26.1-0.20211029081552-c380a0634e9c h1:MIf1yj66j5c+eM6eHGYMjY15UosCRDsaXIVYoioGf5E=
-knative.dev/eventing-kafka v0.26.1-0.20211029081552-c380a0634e9c/go.mod h1:Vz6hgwmnzvwOtOMsbqrqY8UoteyiwbRtlLV/POAdNLg=
-knative.dev/hack v0.0.0-20210806075220-815cd312d65c/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211018110626-47ac3b032e60/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211026141922-a71c865b5f66/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211027200727-f1228dd5e3e6/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e h1:0Hw2xdWYbcs2JRJnOLzAVh7APOtgro7gSno0228mnDg=
-knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack/schema v0.0.0-20211026141922-a71c865b5f66/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/pkg v0.0.0-20210914164111-4857ab6939e3/go.mod h1:jMSqkNMsrzuy+XR4Yr/BMy7SDVbUOl3KKB6+5MR+ZU8=
-knative.dev/pkg v0.0.0-20211018141937-a34efd6b409d/go.mod h1:/d1+Wklc8blR39M4oEBu+GeLav4vsiiJwq61tJay7mc=
-knative.dev/pkg v0.0.0-20211027105800-3b33e02e5b9c/go.mod h1:xtF+ujTgEybZxA+e7NwKgHR8xMjYzHQUBxyaOUEFxwY=
-knative.dev/pkg v0.0.0-20211027171921-f7b70f5ce303/go.mod h1:xtF+ujTgEybZxA+e7NwKgHR8xMjYzHQUBxyaOUEFxwY=
-knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40 h1:eOOOqcj+IHIp5te9GGZsrq6vCXUv2y5JItRT3nXHs20=
-knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40/go.mod h1:HyEqMTLzT2hGCisGaRMGzEE8b4Ym+7mhmVx5ygiX+fY=
-knative.dev/reconciler-test v0.0.0-20211018075026-816dbd1c57f7/go.mod h1:6yDmb26SINSmgw6wVy9qQwgRMewiW8ddkkwGLR0ZvOY=
-knative.dev/reconciler-test v0.0.0-20211019133535-04ea7ddac7e3/go.mod h1:ltMDlEIFo5/9ES92BhIWQkdZaHRnGGJvXcYlvPxZj5s=
-knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c h1:SQ5ag46cbR0ZCJ49zc4F/+hGs0JdudsLaLpt6vSSX+E=
-knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c/go.mod h1:PIwtS0at/rtnsNCxcZUAZUmWKKasTUYJ/w1Mk94bOdA=
+knative.dev/control-protocol v0.0.0-20211101215039-a32569595694/go.mod h1:HlIoGfErGnSVKN6NI3iG5gXxCo9EcdV8sCDGW2IObJo=
+knative.dev/eventing v0.27.0 h1:P9rqKvsCeFb8sEhfK32WFTG7awdOAa7XO6p3umSq/wU=
+knative.dev/eventing v0.27.0/go.mod h1:4ppWQEQ/2B66/YFENDmV1Gjxs4meLpz6UTUgLkkINt4=
+knative.dev/eventing-kafka v0.27.0 h1:mfbQbK+g9PiBHL9FoLTNckQRvY4EzMfmR2z25zTKJeU=
+knative.dev/eventing-kafka v0.27.0/go.mod h1:bt9HdIsI9RMgiWfcsrDT2vH8ADGY+ehN4a6Bj4QH5Jo=
+knative.dev/hack v0.0.0-20211101195839-11d193bf617b h1:DaW1iliZlBAwq/I8gTqVu8UnfGxyb5yR7CDsJi5jyWk=
+knative.dev/hack v0.0.0-20211101195839-11d193bf617b/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/hack/schema v0.0.0-20211101195839-11d193bf617b/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
+knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc h1:ldjbTpoMXUTgzw0IJsAFLdyA6/6QYRvz8IGPZOknEDg=
+knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc/go.mod h1:SkfDk9bWIiNZD7XtILGkG7AKVyF/M6M0bGxLgl0SYL8=
+knative.dev/reconciler-test v0.0.0-20211101214439-9839937c9b13 h1:PgMCfHECOrYopm3P9s2Xq0Hl/MeBI0RoZ/dBH0a3X2g=
+knative.dev/reconciler-test v0.0.0-20211101214439-9839937c9b13/go.mod h1:gTsbLk496j/M9xqMpx/liyCQ0X3bwDpRtcs2Zzws364=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
 modernc.org/golex v1.0.0/go.mod h1:b/QX9oBD/LhixY6NDh+IdGv17hgB+51fET1i2kPSmvk=
 modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k=
diff --git a/vendor/knative.dev/hack/release.sh b/vendor/knative.dev/hack/release.sh
index 80b035fb4..0f7c47324 100644
--- a/vendor/knative.dev/hack/release.sh
+++ b/vendor/knative.dev/hack/release.sh
@@ -131,10 +131,18 @@ function master_version() {
   echo "${tokens[0]}.${tokens[1]}"
 }
 
+# Return the minor version of a release.
+# For example, "v0.2.1" returns "2"
+# Parameters: $1 - release version label.
+function minor_version() {
+  local tokens=(${1//\./ })
+  echo "${tokens[1]}"
+}
+
 # Return the release build number of a release.
 # For example, "v0.2.1" returns "1".
 # Parameters: $1 - release version label.
-function release_build_number() {
+function patch_version() {
   local tokens=(${1//\./ })
   echo "${tokens[2]}"
 }
@@ -171,7 +179,10 @@ function prepare_auto_release() {
   PUBLISH_RELEASE=1
 
   git fetch --all || abort "error fetching branches/tags from remote"
-  local tags="$(git tag | cut -d 'v' -f2 | cut -d '.' -f1-2 | sort -V | uniq)"
+  # Support two different formats for tags
+  # - knative-v1.0.0
+  # - v1.0.0
+  local tags="$(git tag | cut -d '-' -f2 | cut -d 'v' -f2 | cut -d '.' -f1-2 | sort -V | uniq)"
   local branches="$( { (git branch -r | grep upstream/release-) ; (git branch | grep release-); } | cut -d '-' -f2 | sort -V | uniq)"
 
   echo "Versions released (from tags): [" "${tags}" "]"
@@ -210,7 +221,10 @@ function prepare_dot_release() {
   git fetch --all || abort "error fetching branches/tags from remote"
   # List latest release
   local releases # don't combine with the line below, or $? will be 0
-  releases="$(hub_tool release)"
+  # Support tags in two formats
+  # - knative-v1.0.0
+  # - v1.0.0
+  releases="$(hub_tool release | cut -d '-' -f2)"
   echo "Current releases are: ${releases}"
   [[ $? -eq 0 ]] || abort "cannot list releases"
   # If --release-branch passed, restrict to that release
@@ -234,7 +248,9 @@ function prepare_dot_release() {
   [[ -n "${major_minor_version}" ]] || abort "cannot get release major/minor version"
   # Ensure there are new commits in the branch, otherwise we don't create a new release
   setup_branch
-  local last_release_commit="$(git rev-list -n 1 "${last_version}")"
+  # Use the original tag (ie. potentially with a knative- prefix) when determining the last version commit sha
+  local github_tag="$(hub_tool release | grep "${last_version}")"
+  local last_release_commit="$(git rev-list -n 1 "${github_tag}")"
   local release_branch_commit="$(git rev-list -n 1 upstream/"${RELEASE_BRANCH}")"
   [[ -n "${last_release_commit}" ]] || abort "cannot get last release commit"
   [[ -n "${release_branch_commit}" ]] || abort "cannot get release branch last commit"
@@ -246,13 +262,13 @@ function prepare_dot_release() {
     exit 0
   fi
   # Create new release version number
-  local last_build="$(release_build_number "${last_version}")"
+  local last_build="$(patch_version "${last_version}")"
   RELEASE_VERSION="${major_minor_version}.$(( last_build + 1 ))"
   echo "Will create release ${RELEASE_VERSION} at commit ${release_branch_commit}"
   # If --release-notes not used, copy from the latest release
   if [[ -z "${RELEASE_NOTES}" ]]; then
     RELEASE_NOTES="$(mktemp)"
-    hub_tool release show -f "%b" "${last_version}" > "${RELEASE_NOTES}"
+    hub_tool release show -f "%b" "${github_tag}" > "${RELEASE_NOTES}"
     echo "Release notes from ${last_version} copied to ${RELEASE_NOTES}"
   fi
 }
@@ -595,6 +611,8 @@ function publish_to_github() {
   local description="$(mktemp)"
   local attachments_dir="$(mktemp -d)"
   local commitish=""
+  local github_tag="knative-${TAG}"
+
   # Copy files to a separate dir
   for artifact in $@; do
     cp ${artifact} "${attachments_dir}"/
@@ -604,8 +622,22 @@ function publish_to_github() {
   if [[ -n "${RELEASE_NOTES}" ]]; then
     cat "${RELEASE_NOTES}" >> "${description}"
   fi
-  git tag -a "${TAG}" -m "${title}"
-  git_push tag "${TAG}"
+  git tag -a "${github_tag}" -m "${title}"
+  git_push tag "${github_tag}"
+
+  # Include a tag for the go module version
+  #
+  # v1.0.0 = v0.27.0
+  # v1.0.1 = v0.27.1
+  # v1.1.1 = v0.28.1
+  #
+  # See: https://github.com/knative/hack/pull/97
+  if [[ "$TAG" == "v1"* ]]; then
+    local release_minor=$(minor_version $TAG)
+    local go_module_version="v0.$(( release_minor + 27 )).$(patch_version $TAG)"
+    git tag -a "${go_module_version}" -m "${title}"
+    git_push tag "${go_module_version}"
+  fi
 
   [[ -n "${RELEASE_BRANCH}" ]] && commitish="--commitish=${RELEASE_BRANCH}"
   for i in {2..0}; do
@@ -613,7 +645,7 @@ function publish_to_github() {
         ${attachments[@]} \
         --file="${description}" \
         "${commitish}" \
-        "${TAG}" && return 0
+        "${github_tag}" && return 0
     if [[ "${i}" -gt 0 ]]; then
       echo "Error publishing the release, retrying in 15s..."
       sleep 15
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 2377d7cd1..90f9c9146 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1031,7 +1031,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.26.1-0.20211029085753-34c08a406eca
+# knative.dev/eventing v0.27.0
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1140,7 +1140,7 @@ knative.dev/eventing/test/test_images/performance
 knative.dev/eventing/test/test_images/print
 knative.dev/eventing/test/test_images/recordevents
 knative.dev/eventing/test/test_images/request-sender
-# knative.dev/eventing-kafka v0.26.1-0.20211029081552-c380a0634e9c
+# knative.dev/eventing-kafka v0.27.0
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1190,10 +1190,10 @@ 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
+# knative.dev/hack v0.0.0-20211101195839-11d193bf617b
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40
+# knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck
@@ -1286,7 +1286,7 @@ knative.dev/pkg/webhook/json
 knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c
+# knative.dev/reconciler-test v0.0.0-20211101214439-9839937c9b13
 ## explicit
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index aed4d72f6..b913ddc82 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -1887,7 +1887,7 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 			KafkaDefaultTopicDetails:     defaultTopicDetail,
 			KafkaDefaultTopicDetailsLock: sync.RWMutex{},
 			ConfigMapLister:              listers.GetConfigMapLister(),
-			NewKafkaClusterAdmin: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
 					ExpectedTopicName:   fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName),
 					ExpectedTopicDetail: expectedTopicDetail,
diff --git a/control-plane/pkg/reconciler/kafka/topic_test.go b/control-plane/pkg/reconciler/kafka/topic_test.go
index 0fb3a4a38..7694d7cfe 100644
--- a/control-plane/pkg/reconciler/kafka/topic_test.go
+++ b/control-plane/pkg/reconciler/kafka/topic_test.go
@@ -312,7 +312,7 @@ func TestCreateTopicTopicAlreadyExists(t *testing.T) {
 	assert.Nil(t, err, "expected nil error on topic already exists")
 }
 
-func TestNewClusterAdminFuncIsTopicPresent(t *testing.T) {
+func TestNewClusterAdminClientFuncIsTopicPresent(t *testing.T) {
 	tests := []struct {
 		name         string
 		clusterAdmin sarama.ClusterAdmin
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 6629ad98b..787d352df 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -1091,7 +1091,7 @@ func useTable(t *testing.T, table TableTest, configs *broker.Configs) {
 				ReceiverLabel:               base.SinkReceiverLabel,
 			},
 			ConfigMapLister: listers.GetConfigMapLister(),
-			NewKafkaClusterAdmin: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
 					ExpectedTopicName:                      expectedTopicName,
 					ExpectedTopicDetail:                    expectedTopicDetail,
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 1c3fc7b3d..2aa8128a6 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -457,7 +457,7 @@ func useTable(t *testing.T, table TableTest, configs broker.Configs) {
 				DispatcherLabel:             base.SourceDispatcherLabel,
 			},
 			Env: &configs.Env,
-			NewKafkaClusterAdmin: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
 					ExpectedTopicName:                      "",
 					ExpectedTopicDetail:                    sarama.TopicDetail{},