Repository: knative-sandbox/eventing-kafka-broker
Pull request #63 (2020-07-27 17:38:59)

Create a new ClusterAdmin at each loop
Fix `sarama` issue: https://github.com/Shopify/sarama/issues/1565 (in our case is `ClusterAdmin`, we have the same error though)

## Proposed Changes

- Create a new ClusterAdmin at each loop


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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/63/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1287804754663051264/artifacts/line-cov.html#file0) | 90.8% | 91.4% | 0.5
[control-plane/pkg/reconciler/broker/topic.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/63/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1287804754663051264/artifacts/line-cov.html#file2) | 100.0% | 91.7% | -8.3

knative-prow-robot: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 06495adc48486a0f120ef54ddc3c2acf8866405e | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/63/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1287804754658856960) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<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-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/63/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1287812555917496321/artifacts/line-cov.html#file0) | 90.8% | 91.4% | 0.5
[control-plane/pkg/reconciler/broker/topic.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/63/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1287812555917496321/artifacts/line-cov.html#file2) | 100.0% | 91.7% | -8.3

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/63#" title="Author self-approved">pierDipi</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/63#pullrequestreview-456379021" title="Approved">slinkydeveloper</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/master/OWNERS)~~ [slinkydeveloper]

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/.gitignore b/.gitignore
index 19cb6104e..5125ea810 100644
--- a/.gitignore
+++ b/.gitignore
@@ -27,4 +27,7 @@
 **/build/
 
 ### VS Code ###
-**/.vscode/
\ No newline at end of file
+**/.vscode/
+
+## Ignore released file
+eventing-kafka-broker.yaml
\ No newline at end of file
diff --git a/control-plane/config/100-kafka-broker-configmap.yaml b/control-plane/config/100-kafka-broker-configmap.yaml
index 67b1f7fd6..bdf1aeba4 100644
--- a/control-plane/config/100-kafka-broker-configmap.yaml
+++ b/control-plane/config/100-kafka-broker-configmap.yaml
@@ -1,3 +1,19 @@
+---
+
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 apiVersion: v1
 kind: ConfigMap
 metadata:
diff --git a/control-plane/config/200-controller-cluster-role.yaml b/control-plane/config/200-controller-cluster-role.yaml
index 2b455f826..16d18aa0e 100644
--- a/control-plane/config/200-controller-cluster-role.yaml
+++ b/control-plane/config/200-controller-cluster-role.yaml
@@ -1,3 +1,19 @@
+---
+
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
 metadata:
diff --git a/control-plane/config/200-controller-service-account.yaml b/control-plane/config/200-controller-service-account.yaml
index 7de71790f..035854ac2 100644
--- a/control-plane/config/200-controller-service-account.yaml
+++ b/control-plane/config/200-controller-service-account.yaml
@@ -1,3 +1,19 @@
+---
+
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 apiVersion: v1
 kind: ServiceAccount
 metadata:
diff --git a/control-plane/config/201-controller-cluster-role-binding.yaml b/control-plane/config/201-controller-cluster-role-binding.yaml
index 10360de95..e0a26dd17 100644
--- a/control-plane/config/201-controller-cluster-role-binding.yaml
+++ b/control-plane/config/201-controller-cluster-role-binding.yaml
@@ -1,3 +1,19 @@
+---
+
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
@@ -12,4 +28,3 @@ roleRef:
   kind: ClusterRole
   name: kafka-broker-controller
   apiGroup: rbac.authorization.k8s.io
-
diff --git a/control-plane/config/500-controller.yaml b/control-plane/config/500-controller.yaml
index 2909d8ee3..4f09c5165 100644
--- a/control-plane/config/500-controller.yaml
+++ b/control-plane/config/500-controller.yaml
@@ -1,3 +1,19 @@
+---
+
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 apiVersion: apps/v1
 kind: Deployment
 metadata:
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index a5991c945..d9a896b10 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -22,7 +22,6 @@ import (
 	"math"
 	"strings"
 	"sync"
-	"time"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
@@ -53,10 +52,16 @@ type Reconciler struct {
 
 	Resolver *resolver.URIResolver
 
-	KafkaClusterAdmin            sarama.ClusterAdmin
-	KafkaClusterAdminLock        sync.RWMutex
+	// TODO these configurations should live in each Broker configuration, so that we don't assume each
+	// 	Broker object use the same Kafka cluster
 	KafkaDefaultTopicDetails     sarama.TopicDetail
 	KafkaDefaultTopicDetailsLock sync.RWMutex
+	bootstrapServers             []string
+	bootstrapServersLock         sync.RWMutex
+
+	// NewClusterAdmin 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.
+	NewClusterAdmin func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error)
 
 	Configs *Configs
 }
@@ -288,38 +293,30 @@ func (r *Reconciler) ConfigMapUpdated(ctx context.Context) func(configMap *corev
 		topicDetail.ReplicationFactor = int16(replicationFactor)
 
 		r.SetDefaultTopicDetails(topicDetail)
-
-		if err := r.SetBootstrapServers(bootstrapServers); err != nil {
-			logger.Error("Failed to set bootstrap servers", zap.Error(err))
-		}
+		r.SetBootstrapServers(bootstrapServers)
 	}
 }
 
 // SetBootstrapServers change kafka bootstrap brokers addresses.
 // servers: a comma separated list of brokers to connect to.
-func (r *Reconciler) SetBootstrapServers(servers string) error {
+func (r *Reconciler) SetBootstrapServers(servers string) {
 	addrs := strings.Split(servers, ",")
 
+	r.bootstrapServersLock.Lock()
+	r.bootstrapServers = addrs
+	r.bootstrapServersLock.Unlock()
+}
+
+func (r *Reconciler) getKafkaClusterAdmin(bootstrapServers []string) (sarama.ClusterAdmin, error) {
 	config := sarama.NewConfig()
 	config.Version = sarama.MaxVersion
-	config.Net.KeepAlive = time.Second * 60
-	config.Metadata.RefreshFrequency = time.Minute
 
-	kafkaClusterAdmin, err := NewClusterAdmin(addrs, config)
+	kafkaClusterAdmin, err := r.NewClusterAdmin(bootstrapServers, config)
 	if err != nil {
-		return fmt.Errorf("failed to create kafka cluster admin: %w", err)
+		return nil, fmt.Errorf("failed to create cluster admin: %w", err)
 	}
 
-	r.KafkaClusterAdminLock.Lock()
-	oldKafkaClusterAdmin := r.KafkaClusterAdmin
-	r.KafkaClusterAdmin = kafkaClusterAdmin
-	r.KafkaClusterAdminLock.Unlock()
-
-	if oldKafkaClusterAdmin != nil {
-		_ = oldKafkaClusterAdmin.Close()
-	}
-
-	return nil
+	return kafkaClusterAdmin, nil
 }
 
 func (r *Reconciler) SetDefaultTopicDetails(topicDetail sarama.TopicDetail) {
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index f8672150c..7e10c5486 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -46,8 +46,6 @@ const (
 	DefaultReplicationFactor = 1
 )
 
-var NewClusterAdmin = sarama.NewClusterAdmin
-
 func NewController(ctx context.Context, watcher configmap.Watcher, configs *Configs) *controller.Impl {
 
 	eventing.RegisterAlternateBrokerConditionSet(ConditionSet)
@@ -61,6 +59,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
 			SystemNamespace:             configs.SystemNamespace,
 		},
+		NewClusterAdmin: sarama.NewClusterAdmin,
 		KafkaDefaultTopicDetails: sarama.TopicDetail{
 			NumPartitions:     DefaultNumPartitions,
 			ReplicationFactor: DefaultReplicationFactor,
@@ -69,7 +68,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 	}
 
 	if configs.BootstrapServers != "" {
-		_ = reconciler.SetBootstrapServers(configs.BootstrapServers)
+		reconciler.SetBootstrapServers(configs.BootstrapServers)
 	}
 
 	impl := brokerreconciler.NewImpl(ctx, reconciler, kafka.BrokerClass)
diff --git a/control-plane/pkg/reconciler/broker/topic.go b/control-plane/pkg/reconciler/broker/topic.go
index 9c6dabe8c..88225dee4 100644
--- a/control-plane/pkg/reconciler/broker/topic.go
+++ b/control-plane/pkg/reconciler/broker/topic.go
@@ -24,13 +24,18 @@ import (
 )
 
 func (r *Reconciler) CreateTopic(broker *eventing.Broker) (string, error) {
+	r.bootstrapServersLock.RLock()
+	defer r.bootstrapServersLock.RUnlock()
+
 	topic := Topic(broker)
 	topicDetail := r.topicDetailFromBrokerConfig(broker)
 
-	r.KafkaClusterAdminLock.Lock()
-	defer r.KafkaClusterAdminLock.Unlock()
+	kafkaClusterAdmin, err := r.getKafkaClusterAdmin(r.bootstrapServers)
+	if err != nil {
+		return "", err
+	}
 
-	createTopicError := r.KafkaClusterAdmin.CreateTopic(topic, topicDetail, true)
+	createTopicError := kafkaClusterAdmin.CreateTopic(topic, topicDetail, true)
 	if err, ok := createTopicError.(*sarama.TopicError); ok && err.Err == sarama.ErrTopicAlreadyExists {
 		return topic, nil
 	}
@@ -39,13 +44,17 @@ func (r *Reconciler) CreateTopic(broker *eventing.Broker) (string, error) {
 }
 
 func (r *Reconciler) deleteTopic(broker *eventing.Broker) (string, error) {
-
-	r.KafkaClusterAdminLock.RLock()
-	defer r.KafkaClusterAdminLock.RUnlock()
+	r.bootstrapServersLock.RLock()
+	defer r.bootstrapServersLock.RUnlock()
 
 	topic := Topic(broker)
 
-	err := r.KafkaClusterAdmin.DeleteTopic(topic)
+	kafkaClusterAdmin, err := r.getKafkaClusterAdmin(r.bootstrapServers)
+	if err != nil {
+		return "", err
+	}
+
+	err = kafkaClusterAdmin.DeleteTopic(topic)
 	if sarama.ErrUnknownTopicOrPartition == err {
 		return topic, nil
 	}
diff --git a/data-plane/config/100-config-brokers-triggers.yaml b/data-plane/config/100-config-brokers-triggers.yaml
index 16167abdf..842187d11 100644
--- a/data-plane/config/100-config-brokers-triggers.yaml
+++ b/data-plane/config/100-config-brokers-triggers.yaml
@@ -1,3 +1,19 @@
+---
+
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 apiVersion: v1
 kind: ConfigMap
 metadata:
@@ -7,5 +23,3 @@ metadata:
     eventing.knative.dev/release: devel
 binaryData:
   data: ""
-
-
diff --git a/data-plane/config/100-config-kafka-broker-dispatcher.yaml b/data-plane/config/100-config-kafka-broker-dispatcher.yaml
index 16838f15a..65846f386 100644
--- a/data-plane/config/100-config-kafka-broker-dispatcher.yaml
+++ b/data-plane/config/100-config-kafka-broker-dispatcher.yaml
@@ -1,3 +1,19 @@
+---
+
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 apiVersion: v1
 kind: ConfigMap
 metadata:
diff --git a/data-plane/config/100-config-kafka-broker-receiver.yaml b/data-plane/config/100-config-kafka-broker-receiver.yaml
index a39951db3..be6600ca0 100644
--- a/data-plane/config/100-config-kafka-broker-receiver.yaml
+++ b/data-plane/config/100-config-kafka-broker-receiver.yaml
@@ -1,3 +1,19 @@
+---
+
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 apiVersion: v1
 kind: ConfigMap
 metadata:
diff --git a/data-plane/config/100-config-logging.yaml b/data-plane/config/100-config-logging.yaml
index b7011b3ed..5c8b04f22 100644
--- a/data-plane/config/100-config-logging.yaml
+++ b/data-plane/config/100-config-logging.yaml
@@ -1,3 +1,19 @@
+---
+
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 apiVersion: v1
 kind: ConfigMap
 metadata:
@@ -12,4 +28,5 @@ data:
       <root level="INFO">
         <appender-ref ref="jsonConsoleAppender"/>
       </root>
-    </configuration>
\ No newline at end of file
+    </configuration>
+
diff --git a/data-plane/config/template/500-dispatcher.yaml b/data-plane/config/template/500-dispatcher.yaml
index c0ffc9942..6932df890 100644
--- a/data-plane/config/template/500-dispatcher.yaml
+++ b/data-plane/config/template/500-dispatcher.yaml
@@ -1,3 +1,19 @@
+---
+
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 apiVersion: apps/v1
 kind: Deployment
 metadata:
diff --git a/data-plane/config/template/500-receiver.yaml b/data-plane/config/template/500-receiver.yaml
index f876fb9fa..31e6cf32b 100644
--- a/data-plane/config/template/500-receiver.yaml
+++ b/data-plane/config/template/500-receiver.yaml
@@ -1,3 +1,19 @@
+---
+
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -66,7 +82,6 @@ spec:
             periodSeconds: 3
             successThreshold: 1
             timeoutSeconds: 1
-          # TODO check kafka reachability
           readinessProbe:
             failureThreshold: 3
             httpGet:
@@ -97,6 +112,7 @@ spec:
             name: kafka-broker-config-logging
       restartPolicy: Always
 ---
+
 apiVersion: v1
 kind: Service
 metadata:
@@ -111,3 +127,4 @@ spec:
       protocol: TCP
       targetPort: 8080
   type: NodePort
+---
diff --git a/hack/release.sh b/hack/release.sh
new file mode 100644
index 000000000..1d442b210
--- /dev/null
+++ b/hack/release.sh
@@ -0,0 +1,34 @@
+#!/usr/bin/env bash
+
+# Copyright 2020 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.
+
+set -o errexit
+set -o nounset
+set -o pipefail
+
+source $(dirname $0)/../vendor/knative.dev/test-infra/scripts/release.sh
+source $(dirname $0)/../test/data-plane/library.sh
+source $(dirname $0)/../test/control-plane-plane/library.sh
+
+export EVENTING_KAFKA_BROKER_ARTIFACT="eventing-kafka-broker.yaml"
+
+function build_release() {
+  data_plane_setup && control_plane_setup
+  if [ -n $? ]; then
+    export ARTIFACTS_TO_PUBLISH=(${EVENTING_KAFKA_BROKER_ARTIFACT})
+  fi
+}
+
+main $@


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index a6fa65e10..aeacf795a 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -960,14 +960,6 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 			onDeleteTopicError = want.(error)
 		}
 
-		clusterAdmin := &MockKafkaClusterAdmin{
-			ExpectedTopicName:   fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName),
-			ExpectedTopicDetail: defaultTopicDetail,
-			ErrorOnCreateTopic:  onCreateTopicError,
-			ErrorOnDeleteTopic:  onDeleteTopicError,
-			T:                   t,
-		}
-
 		reconciler := &Reconciler{
 			Reconciler: &base.Reconciler{
 				KubeClient:                  kubeclient.Get(ctx),
@@ -977,7 +969,15 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 				DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
 				SystemNamespace:             configs.SystemNamespace,
 			},
-			KafkaClusterAdmin:            clusterAdmin,
+			NewClusterAdmin: func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &MockKafkaClusterAdmin{
+					ExpectedTopicName:   fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName),
+					ExpectedTopicDetail: defaultTopicDetail,
+					ErrorOnCreateTopic:  onCreateTopicError,
+					ErrorOnDeleteTopic:  onDeleteTopicError,
+					T:                   t,
+				}, nil
+			},
 			KafkaDefaultTopicDetails:     defaultTopicDetail,
 			KafkaDefaultTopicDetailsLock: sync.RWMutex{},
 			Configs:                      configs,
@@ -1018,10 +1018,6 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 
 func TestConfigMapUpdate(t *testing.T) {
 
-	NewClusterAdmin = func(addrs []string, conf *sarama.Config) (sarama.ClusterAdmin, error) {
-		return MockKafkaClusterAdmin{}, nil
-	}
-
 	cm := corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "cmname",
@@ -1044,7 +1040,6 @@ func TestConfigMapUpdate(t *testing.T) {
 		NumPartitions:     42,
 		ReplicationFactor: 3,
 	})
-	assert.NotNil(t, reconciler.KafkaClusterAdmin)
 }
 
 func patchFinalizers() clientgotesting.PatchActionImpl {
diff --git a/control-plane/pkg/reconciler/broker/controller_test.go b/control-plane/pkg/reconciler/broker/controller_test.go
index fe36b290b..dc299faf6 100644
--- a/control-plane/pkg/reconciler/broker/controller_test.go
+++ b/control-plane/pkg/reconciler/broker/controller_test.go
@@ -19,7 +19,6 @@ package broker
 import (
 	"testing"
 
-	"github.com/Shopify/sarama"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -37,10 +36,6 @@ import (
 func TestNewController(t *testing.T) {
 	ctx, _ := reconcilertesting.SetupFakeContext(t)
 
-	NewClusterAdmin = func(addrs []string, conf *sarama.Config) (sarama.ClusterAdmin, error) {
-		return nil, nil
-	}
-
 	configs := &Configs{
 		EnvConfigs: EnvConfigs{
 			SystemNamespace:      "cm",
diff --git a/control-plane/pkg/reconciler/broker/topic_test.go b/control-plane/pkg/reconciler/broker/topic_test.go
index effdb8e60..f63f5882b 100644
--- a/control-plane/pkg/reconciler/broker/topic_test.go
+++ b/control-plane/pkg/reconciler/broker/topic_test.go
@@ -40,14 +40,16 @@ func TestCreateTopicTopicAlreadyExists(t *testing.T) {
 	errMsg := "topic already exists"
 
 	r := broker.Reconciler{
-		KafkaClusterAdmin: reconcilertesting.MockKafkaClusterAdmin{
-			ExpectedTopicName:   topic,
-			ExpectedTopicDetail: sarama.TopicDetail{},
-			ErrorOnCreateTopic: &sarama.TopicError{
-				Err:    sarama.ErrTopicAlreadyExists,
-				ErrMsg: &errMsg,
-			},
-			T: t,
+		NewClusterAdmin: func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
+			return reconcilertesting.MockKafkaClusterAdmin{
+				ExpectedTopicName:   topic,
+				ExpectedTopicDetail: sarama.TopicDetail{},
+				ErrorOnCreateTopic: &sarama.TopicError{
+					Err:    sarama.ErrTopicAlreadyExists,
+					ErrMsg: &errMsg,
+				},
+				T: t,
+			}, nil
 		},
 	}
 
diff --git a/test/control-plane/library.sh b/test/control-plane/library.sh
index 98e9ac237..c01d7ba72 100644
--- a/test/control-plane/library.sh
+++ b/test/control-plane/library.sh
@@ -17,12 +17,17 @@
 CONTROL_PLANE_DIR=control-plane
 CONTROL_PLANE_CONFIG_DIR=${CONTROL_PLANE_DIR}/config
 
-function control_plane_setup() {
-  ko apply --strict -f "${CONTROL_PLANE_CONFIG_DIR}"
-  return $?
-}
+# Update release labels if this is a tagged release
+if [[ -n "${TAG}" ]]; then
+  echo "Tagged release, updating release labels to eventing.knative.dev/release: \"${TAG}\""
+  LABEL_YAML_CMD=(sed -e "s|eventing.knative.dev/release: devel|eventing.knative.dev/release: \"${TAG}\"|")
+else
+  echo "Untagged release, will NOT update release labels"
+  LABEL_YAML_CMD=(cat)
+fi
 
-function control_plane_teardown() {
-  ko apply --strict -f "${CONTROL_PLANE_CONFIG_DIR}"
+# Note: do not change this function name, it's used during releases.
+function control_plane_setup() {
+  ko resolve --strict -f "${CONTROL_PLANE_CONFIG_DIR}" | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_BROKER_ARTIFACT}"
   return $?
 }
diff --git a/test/data-plane/library.sh b/test/data-plane/library.sh
index 41565279d..32fa1ab94 100755
--- a/test/data-plane/library.sh
+++ b/test/data-plane/library.sh
@@ -22,7 +22,7 @@
 
 readonly WITH_KIND=${WITH_KIND:-false}
 readonly SKIP_PUSH=${SKIP_PUSH:-false}
-readonly UUID=${UUID:-latest}
+readonly UUID=${UUID:-${TAG:-latest}}
 
 readonly DATA_PLANE_DIR=data-plane
 readonly DATA_PLANE_CONFIG_DIR=${DATA_PLANE_DIR}/config
@@ -41,6 +41,15 @@ readonly RECEIVER_DIRECTORY=receiver
 readonly DISPATCHER_JAR="dispatcher-1.0-SNAPSHOT.jar"
 readonly DISPATCHER_DIRECTORY=dispatcher
 
+# Update release labels if this is a tagged release
+if [[ -n "${TAG}" ]]; then
+  echo "Tagged release, updating release labels to eventing.knative.dev/release: \"${TAG}\""
+  LABEL_YAML_CMD=(sed -e "s|eventing.knative.dev/release: devel|eventing.knative.dev/release: \"${TAG}\"|")
+else
+  echo "Untagged release, will NOT update release labels"
+  LABEL_YAML_CMD=(cat)
+fi
+
 # Checks whether the given function exists.
 function function_exists() {
   [[ "$(type -t $1)" == "function" ]]
@@ -113,13 +122,13 @@ function k8s() {
   echo "dispatcher image ---> ${KNATIVE_KAFKA_BROKER_DISPATCHER_IMAGE}"
   echo "receiver image   ---> ${KNATIVE_KAFKA_BROKER_RECEIVER_IMAGE}"
 
-  kubectl "$@" -f ${DATA_PLANE_CONFIG_DIR}
+  ko resolve ${KO_FLAGS} --strict -f ${DATA_PLANE_CONFIG_DIR} | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_BROKER_ARTIFACT}"
 
   sed "s|\${KNATIVE_KAFKA_BROKER_DISPATCHER_IMAGE}|${KNATIVE_KAFKA_BROKER_DISPATCHER_IMAGE}|g" ${DISPATCHER_TEMPLATE_FILE} |
-    kubectl "$@" -f - || fail_test "Failed to $@ to ${DISPATCHER_TEMPLATE_FILE}"
+    "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_BROKER_ARTIFACT}" || fail_test "Failed to append ${DISPATCHER_TEMPLATE_FILE}"
 
   sed "s|\${KNATIVE_KAFKA_BROKER_RECEIVER_IMAGE}|${KNATIVE_KAFKA_BROKER_RECEIVER_IMAGE}|g" ${RECEIVER_TEMPLATE_FILE} |
-    kubectl "$@" -f - || fail_test "Failed to $@ to ${RECEIVER_TEMPLATE_FILE}"
+    "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_BROKER_ARTIFACT}" || fail_test "Failed to append ${RECEIVER_TEMPLATE_FILE}"
 }
 
 function data_plane_unit_tests() {
@@ -134,12 +143,8 @@ function data_plane_build_tests() {
   return 0
 }
 
+# Note: do not change this function name, it's used during releases.
 function data_plane_setup() {
-  data_plane_build_push && k8s apply
-  return $?
-}
-
-function data_plane_teardown() {
-  k8s delete --ignore-not-found
+  data_plane_build_push && k8s
   return $?
 }
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index 64812db6e..19d12066f 100644
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -28,6 +28,8 @@ readonly EVENTING_CONFIG="./config"
 # Vendored eventing test iamges.
 readonly VENDOR_EVENTING_TEST_IMAGES="vendor/knative.dev/eventing/test/test_images/"
 
+export EVENTING_KAFKA_BROKER_ARTIFACT="eventing-kafka-broker.yaml"
+
 function knative_setup() {
   knative_eventing "apply --strict"
   return $?
@@ -50,9 +52,7 @@ function knative_eventing() {
   else
     fail_test "Not ready for a release"
   fi
-}
 
-function test_setup() {
   # Publish test images.
   echo ">> Publishing test images from eventing"
   # We vendor test image code from eventing, in order to use ko to resolve them into Docker images, the
@@ -65,18 +65,20 @@ function test_setup() {
   # ./test/upload-test-images.sh "test/test_images" e2e || fail_test "Error uploading test images"
 
   ./test/kafka/kafka_setup.sh || fail_test "Failed to set up Kafka cluster"
+}
+
+function test_setup() {
+  [ -f "${EVENTING_KAFKA_BROKER_ARTIFACT}" ] && rm "${EVENTING_KAFKA_BROKER_ARTIFACT}"
 
   header "Data plane setup"
   data_plane_setup || fail_test "Failed to set up data plane components"
 
   header "Control plane setup"
   control_plane_setup || fail_test "Failed to set up control plane components"
+
+  kubectl apply -f "${EVENTING_KAFKA_BROKER_ARTIFACT}" || fail_test "Failed to apply ${EVENTING_KAFKA_BROKER_ARTIFACT}"
 }
 
 function test_teardown() {
-  header "Data plane teardown"
-  data_plane_teardown || fail_test "Failed to tear down data plane components"
-
-  header "Control plane teardown"
-  control_plane_teardown || fail_test "Failed to tear down control plane components"
+  kubectl delete -f "${EVENTING_KAFKA_BROKER_ARTIFACT}" || fail_test "Failed to tear down"
 }
diff --git a/test/presubmit-tests.sh b/test/presubmit-tests.sh
index 2fca07803..0a3af0bc0 100755
--- a/test/presubmit-tests.sh
+++ b/test/presubmit-tests.sh
@@ -25,7 +25,7 @@ export GO111MODULE=on
 export DISABLE_MD_LINTING=1
 
 source $(dirname $0)/../vendor/knative.dev/test-infra/scripts/presubmit-tests.sh
-source $(dirname $0)/data-plane/library.sh
+source $(dirname $0)/e2e-common.sh
 
 function fail_test() {
   header "$1"