Repository: knative-sandbox/eventing-kafka-broker
Pull request #1983 (2022-03-09 08:15:05)

Consolidate Kafka packages into one single package
These 2 packages `pkg/reconciler/kafka` and `pkg/kafka` always
confuse me about their purpose, so I think it makes more sense to
keep one single package for "Kafka" related functions.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

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/kafka/broker_class.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1983/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1501471625994833920/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[control-plane/pkg/kafka/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1983/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1501471625994833920/artifacts/line-cov.html#file2) | Do not exist | 0.0% | 
[control-plane/pkg/kafka/topic.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1983/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1501471625994833920/artifacts/line-cov.html#file3) | Do not exist | 84.8% | 

knative-prow-robot: @pierDipi: 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-go-coverage | c623911cbc128d961978c7e46dffb7a144364a83 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1983/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1501471625994833920) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1983). [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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1983](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (c623911) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/f02d075812d29123a838fb65602055364bd4959f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (f02d075) will **decrease** coverage by `0.01%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983/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/1983?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    #1983      +/-   ##
============================================
- Coverage     66.99%   66.98%   -0.02%     
+ Complexity      633      632       -1     
============================================
  Files           140      140              
  Lines          6560     6560              
  Branches        179      179              
============================================
- Hits           4395     4394       -1     
- Misses         1796     1797       +1     
  Partials        369      369              
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.32% <ø> (-0.05%)` | :arrow_down: |

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/1983?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ol-plane/pkg/apis/eventing/v1/broker\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MS9icm9rZXJfdmFsaWRhdGlvbi5nbw==) | `75.00% <ø> (ø)` | |
| [control-plane/pkg/kafka/broker\_class.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cva2Fma2EvYnJva2VyX2NsYXNzLmdv) | `100.00% <ø> (ø)` | |
| [control-plane/pkg/kafka/client.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cva2Fma2EvY2xpZW50Lmdv) | `0.00% <ø> (ø)` | |
| [control-plane/pkg/kafka/topic.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cva2Fma2EvdG9waWMuZ28=) | `80.68% <ø> (ø)` | |
| [control-plane/pkg/reconciler/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983/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.00% <ø> (ø)` | |
| [control-plane/pkg/reconciler/broker/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvY29udHJvbGxlci5nbw==) | `79.68% <ø> (ø)` | |
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | `71.28% <ø> (ø)` | |
| [...ntrol-plane/pkg/reconciler/channel/v2/channelv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL3YyL2NoYW5uZWx2Mi5nbw==) | `71.95% <ø> (ø)` | |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | `72.78% <ø> (ø)` | |
| [control-plane/pkg/reconciler/source/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvY29udHJvbGxlci5nbw==) | `72.34% <ø> (ø)` | |
| ... and [8 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983/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/1983?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/1983?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [f02d075...c623911](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1983?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).

pierDipi: /cc @devguyio @matzew @aliok 

Patch:
diff --git a/control-plane/pkg/apis/eventing/v1/broker_validation.go b/control-plane/pkg/apis/eventing/v1/broker_validation.go
index d8e0f11dd..fcfd551b8 100644
--- a/control-plane/pkg/apis/eventing/v1/broker_validation.go
+++ b/control-plane/pkg/apis/eventing/v1/broker_validation.go
@@ -29,7 +29,7 @@ import (
 	"knative.dev/pkg/webhook"
 	"knative.dev/pkg/webhook/resourcesemantics/validation"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 )
 
 type BrokerPartial struct {
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 689f57e2c..f56261acb 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -37,11 +37,11 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 )
 
@@ -122,26 +122,10 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		return fmt.Errorf("failed to track secret: %w", err)
 	}
 
-	topicName := kafka.BrokerTopic(TopicPrefix, broker)
-
-	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
-	if err != nil {
-		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin config: %w", err))
-	}
-
-	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, saramaConfig)
+	topic, err := r.reconcileBrokerTopic(broker, securityOption, statusConditionManager, topicConfig, logger)
 	if err != nil {
-		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("cannot obtain Kafka cluster admin, %w", err))
-	}
-	defer kafkaClusterAdminClient.Close()
-
-	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdminClient, logger, topicName, topicConfig)
-	if err != nil {
-		return statusConditionManager.FailedToCreateTopic(topic, err)
+		return err
 	}
-	statusConditionManager.TopicReady(topic)
-
-	logger.Debug("Topic created", zap.Any("topic", topic))
 
 	// Get contract config map.
 	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
@@ -241,6 +225,30 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	return nil
 }
 
+func (r *Reconciler) reconcileBrokerTopic(broker *eventing.Broker, securityOption kafka.ConfigOption, statusConditionManager base.StatusConditionManager, topicConfig *kafka.TopicConfig, logger *zap.Logger) (string, reconciler.Event) {
+
+	topicName := kafka.BrokerTopic(TopicPrefix, broker)
+	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
+	if err != nil {
+		return "", statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin config: %w", err))
+	}
+
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, saramaConfig)
+	if err != nil {
+		return "", statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("cannot obtain Kafka cluster admin, %w", err))
+	}
+	defer kafkaClusterAdminClient.Close()
+
+	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdminClient, logger, topicName, topicConfig)
+	if err != nil {
+		return "", statusConditionManager.FailedToCreateTopic(topic, err)
+	}
+
+	statusConditionManager.TopicReady(topic)
+	logger.Debug("Topic created", zap.Any("topic", topic))
+	return topic, nil
+}
+
 func (r *Reconciler) FinalizeKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
 	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
 		return r.finalizeKind(ctx, broker)
@@ -325,7 +333,19 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 
 	// get security option for Sarama with secret info in it
 	securityOption := security.NewSaramaSecurityOptionFromSecret(secret)
+	if err := r.finalizeBrokerTopic(broker, securityOption, topicConfig, logger); err != nil {
+		return err
+	}
 
+	// TODO(pierDipi) remove after some releases (released in 1.4)
+	if err := r.removeFinalizerCM(ctx, finalizerCM(broker), brokerConfig); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (r *Reconciler) finalizeBrokerTopic(broker *eventing.Broker, securityOption kafka.ConfigOption, topicConfig *kafka.TopicConfig, logger *zap.Logger) reconciler.Event {
 	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 	if err != nil {
 		// even in error case, we return `normal`, since we are fine with leaving the
@@ -347,12 +367,6 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 	}
 
 	logger.Debug("Topic deleted", zap.String("topic", topic))
-
-	// TODO(pierDipi) remove after some releases (released in 1.4)
-	if err := r.removeFinalizerCM(ctx, finalizerCM(broker), brokerConfig); err != nil {
-		return err
-	}
-
 	return nil
 }
 
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index d488bee56..68c5b9116 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -40,9 +40,9 @@ import (
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 )
 
 const (
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index b79fa344e..da58834d6 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -45,11 +45,11 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 )
 
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2.go b/control-plane/pkg/reconciler/channel/v2/channelv2.go
index 3d0c1f36d..417a2a698 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2.go
@@ -48,10 +48,10 @@ import (
 
 	"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/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 14a96a558..8b7845faf 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -33,11 +33,11 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 )
 
diff --git a/control-plane/pkg/reconciler/source/controller.go b/control-plane/pkg/reconciler/source/controller.go
index f3f0aebc8..33d715ce7 100644
--- a/control-plane/pkg/reconciler/source/controller.go
+++ b/control-plane/pkg/reconciler/source/controller.go
@@ -37,8 +37,8 @@ import (
 	"knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 )
 
 func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env) *controller.Impl {
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index 935fb5f16..dcf977812 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -32,9 +32,9 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 )
 
diff --git a/control-plane/pkg/reconciler/trigger/controller.go b/control-plane/pkg/reconciler/trigger/controller.go
index cee3304f8..32a4b4775 100644
--- a/control-plane/pkg/reconciler/trigger/controller.go
+++ b/control-plane/pkg/reconciler/trigger/controller.go
@@ -40,8 +40,8 @@ import (
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 )
 
 const (
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index a4a8b8b55..7d264c2ae 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -21,9 +21,6 @@ import (
 	"fmt"
 	"strings"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
-
 	"go.uber.org/zap"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -35,6 +32,9 @@ import (
 	"knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
diff --git a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
index 563eb06da..6d1194b0d 100644
--- a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
@@ -30,12 +30,11 @@ import (
 	triggerinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger"
 	triggerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumergroup"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
-
 	consumergroupclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
 	consumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumergroup"
 
 	apiseventing "knative.dev/eventing/pkg/apis/eventing"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
index 127829aac..90ab36c6b 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
@@ -36,8 +36,8 @@ import (
 	internalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned"
 	internalslst "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 )
 
 const (
diff --git a/control-plane/pkg/security/config.go b/control-plane/pkg/security/config.go
index bb54b78fc..22b2f1b45 100644
--- a/control-plane/pkg/security/config.go
+++ b/control-plane/pkg/security/config.go
@@ -25,7 +25,8 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 )
 
 const (
diff --git a/control-plane/pkg/security/secret.go b/control-plane/pkg/security/secret.go
index dcd00af45..4dd8c7d82 100644
--- a/control-plane/pkg/security/secret.go
+++ b/control-plane/pkg/security/secret.go
@@ -23,7 +23,8 @@ import (
 	"strconv"
 
 	"github.com/Shopify/sarama"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 )
 
 const (
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 3d72f195d..ffede864b 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -85,18 +85,18 @@ Lists of 158 third-party dependencies.
      (Apache License, Version 2.0) Netty/Transport/Classes/Epoll (io.netty:netty-transport-classes-epoll:4.1.74.Final - https://netty.io/netty-transport-classes-epoll/)
      (Apache License, Version 2.0) Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.74.Final - https://netty.io/netty-transport-native-epoll/)
      (Apache License, Version 2.0) Netty/Transport/Native/Unix/Common (io.netty:netty-transport-native-unix-common:4.1.74.Final - https://netty.io/netty-transport-native-unix-common/)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-zipkin:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-extension-trace-propagators:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-common:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-logs:1.11.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-metrics:1.11.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-testing:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-trace:1.11.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-semconv:1.11.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:1.12.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:1.12.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:1.12.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-zipkin:1.12.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-extension-trace-propagators:1.12.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk:1.12.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-common:1.12.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-logs:1.12.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-metrics:1.12.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-testing:1.12.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-trace:1.12.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-semconv:1.12.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient (io.prometheus:simpleclient:0.12.0 - http://github.com/prometheus/client_java/simpleclient)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient Common (io.prometheus:simpleclient_common:0.12.0 - http://github.com/prometheus/client_java/simpleclient_common)
      (The Apache Software License, Version 2.0) Prometheus Java Span Context Supplier - Common (io.prometheus:simpleclient_tracer_common:0.12.0 - http://github.com/prometheus/client_java/simpleclient_tracer/simpleclient_tracer_common)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 7414e709a..2bb5decaa 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -44,7 +44,7 @@
     <vertx.version>4.2.5</vertx.version>
     <cloudevents.sdk.version>2.3.0</cloudevents.sdk.version>
     <micrometer.version>1.8.3</micrometer.version>
-    <opentelemetry.version>1.11.0</opentelemetry.version>
+    <opentelemetry.version>1.12.0</opentelemetry.version>
     <jackson.version>2.13.2</jackson.version>
     <protobuf.version>3.19.4</protobuf.version>
     <slf4j.version>1.7.36</slf4j.version>


Test patch:
diff --git a/control-plane/pkg/reconciler/kafka/broker_class.go b/control-plane/pkg/kafka/broker_class.go
similarity index 100%
rename from control-plane/pkg/reconciler/kafka/broker_class.go
rename to control-plane/pkg/kafka/broker_class.go
diff --git a/control-plane/pkg/reconciler/kafka/broker_class_test.go b/control-plane/pkg/kafka/broker_class_test.go
similarity index 100%
rename from control-plane/pkg/reconciler/kafka/broker_class_test.go
rename to control-plane/pkg/kafka/broker_class_test.go
diff --git a/control-plane/pkg/reconciler/kafka/client.go b/control-plane/pkg/kafka/client.go
similarity index 100%
rename from control-plane/pkg/reconciler/kafka/client.go
rename to control-plane/pkg/kafka/client.go
diff --git a/control-plane/pkg/reconciler/kafka/consumer_group.go b/control-plane/pkg/kafka/consumer_group.go
similarity index 100%
rename from control-plane/pkg/reconciler/kafka/consumer_group.go
rename to control-plane/pkg/kafka/consumer_group.go
diff --git a/control-plane/pkg/reconciler/kafka/testing/admin_mock.go b/control-plane/pkg/kafka/testing/admin_mock.go
similarity index 100%
rename from control-plane/pkg/reconciler/kafka/testing/admin_mock.go
rename to control-plane/pkg/kafka/testing/admin_mock.go
diff --git a/control-plane/pkg/reconciler/kafka/testing/client_mock.go b/control-plane/pkg/kafka/testing/client_mock.go
similarity index 100%
rename from control-plane/pkg/reconciler/kafka/testing/client_mock.go
rename to control-plane/pkg/kafka/testing/client_mock.go
diff --git a/control-plane/pkg/reconciler/kafka/topic.go b/control-plane/pkg/kafka/topic.go
similarity index 100%
rename from control-plane/pkg/reconciler/kafka/topic.go
rename to control-plane/pkg/kafka/topic.go
diff --git a/control-plane/pkg/reconciler/kafka/topic_test.go b/control-plane/pkg/kafka/topic_test.go
similarity index 99%
rename from control-plane/pkg/reconciler/kafka/topic_test.go
rename to control-plane/pkg/kafka/topic_test.go
index 67a8c04e9..0d4b6c4c3 100644
--- a/control-plane/pkg/reconciler/kafka/topic_test.go
+++ b/control-plane/pkg/kafka/topic_test.go
@@ -29,7 +29,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 
-	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka/testing"
+	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/testing"
 )
 
 func TestCreateTopic(t *testing.T) {
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index df31de9ad..7d3dc8aaa 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -27,6 +27,8 @@ import (
 	"knative.dev/pkg/network"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/testing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober/probertesting"
 
@@ -56,8 +58,6 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
-	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka/testing"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 )
 
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index a888eee34..21f8c9ced 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -24,7 +24,8 @@ import (
 	"github.com/Shopify/sarama"
 	"knative.dev/pkg/network"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/testing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 
 	corev1 "k8s.io/api/core/v1"
@@ -46,7 +47,6 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
-	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka/testing"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
index 5694a524c..be8ce3afb 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
@@ -34,24 +34,26 @@ import (
 	. "knative.dev/pkg/reconciler/testing"
 	"knative.dev/pkg/system"
 
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	"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/kafka"
+	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/testing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober/probertesting"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
-	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka/testing"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
-	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 
-	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
-	fakeconsumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	fakeeventingkafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
 	messagingv1beta1kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
+
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	fakeconsumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
 )
 
 const (
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index b3b157db5..4ea2cafc3 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -27,6 +27,7 @@ import (
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/testing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober/probertesting"
 
@@ -47,7 +48,6 @@ import (
 	sinkreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka/testing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/sink"
 	reconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/sink"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 1220f782a..d341c1a47 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -35,6 +35,7 @@ import (
 	"knative.dev/pkg/tracker"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/testing"
 
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
@@ -45,7 +46,6 @@ import (
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka/testing"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/source"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 )
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index de4788d39..8e57cb46a 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -35,11 +35,11 @@ import (
 
 	"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/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 )
 
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
index 87e4989d1..8aaf8bd19 100644
--- a/control-plane/pkg/reconciler/testing/objects_channel.go
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -33,9 +33,9 @@ import (
 
 	"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/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 )
 
 const (
diff --git a/control-plane/pkg/reconciler/trigger/controller_test.go b/control-plane/pkg/reconciler/trigger/controller_test.go
index e254cd28c..48491fc93 100644
--- a/control-plane/pkg/reconciler/trigger/controller_test.go
+++ b/control-plane/pkg/reconciler/trigger/controller_test.go
@@ -35,7 +35,7 @@ import (
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 )
 
 func TestNewController(t *testing.T) {
diff --git a/control-plane/pkg/security/secret_test.go b/control-plane/pkg/security/secret_test.go
index 65cb5f982..604702f77 100644
--- a/control-plane/pkg/security/secret_test.go
+++ b/control-plane/pkg/security/secret_test.go
@@ -23,7 +23,7 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 )
 
 func TestNoProtocol(t *testing.T) {
diff --git a/test/e2e/broker_redelivery_test.go b/test/e2e/broker_redelivery_test.go
index a569611f3..937781e2d 100644
--- a/test/e2e/broker_redelivery_test.go
+++ b/test/e2e/broker_redelivery_test.go
@@ -29,7 +29,7 @@ import (
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/resources"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	kafkatesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
diff --git a/test/e2e/broker_sasl_ssl_test.go b/test/e2e/broker_sasl_ssl_test.go
index f51376137..efa3ee71f 100644
--- a/test/e2e/broker_sasl_ssl_test.go
+++ b/test/e2e/broker_sasl_ssl_test.go
@@ -38,7 +38,7 @@ import (
 	"knative.dev/eventing/test/lib/sender"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	. "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
diff --git a/test/e2e/broker_trigger_sink_test.go b/test/e2e/broker_trigger_sink_test.go
index 89fe327b8..27fb01cdd 100644
--- a/test/e2e/broker_trigger_sink_test.go
+++ b/test/e2e/broker_trigger_sink_test.go
@@ -34,7 +34,7 @@ import (
 
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	eventingv1alpha1clientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/eventing/v1alpha1"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/test/pkg/addressable"
 	"knative.dev/eventing-kafka-broker/test/pkg/sink"
 	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
diff --git a/test/e2e/broker_trigger_test.go b/test/e2e/broker_trigger_test.go
index 00ab0bd51..9c1b3198d 100644
--- a/test/e2e/broker_trigger_test.go
+++ b/test/e2e/broker_trigger_test.go
@@ -34,8 +34,8 @@ import (
 	"knative.dev/eventing/test/lib/resources"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 	kafkatest "knative.dev/eventing-kafka-broker/test/pkg/kafka"
 	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
diff --git a/test/e2e/conformance/control_plane_conformance_test.go b/test/e2e/conformance/control_plane_conformance_test.go
index f3e190684..c4950297c 100644
--- a/test/e2e/conformance/control_plane_conformance_test.go
+++ b/test/e2e/conformance/control_plane_conformance_test.go
@@ -26,7 +26,7 @@ import (
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/resources"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
diff --git a/test/e2e/conformance/data_plane_conformance_test.go b/test/e2e/conformance/data_plane_conformance_test.go
index f93f42956..4459b818c 100644
--- a/test/e2e/conformance/data_plane_conformance_test.go
+++ b/test/e2e/conformance/data_plane_conformance_test.go
@@ -34,7 +34,7 @@ import (
 	"knative.dev/eventing/test/lib/sender"
 	pkgtest "knative.dev/pkg/test"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
diff --git a/test/e2e_new/broker_conformance_test.go b/test/e2e_new/broker_conformance_test.go
index decd6d861..c99a223b6 100644
--- a/test/e2e_new/broker_conformance_test.go
+++ b/test/e2e_new/broker_conformance_test.go
@@ -34,7 +34,7 @@ import (
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 )
 
 func TestBrokerConformance(t *testing.T) {
diff --git a/test/e2e_new/broker_delete_config_map_test.go b/test/e2e_new/broker_delete_config_map_test.go
index 0c60b1621..177a28416 100644
--- a/test/e2e_new/broker_delete_config_map_test.go
+++ b/test/e2e_new/broker_delete_config_map_test.go
@@ -23,8 +23,6 @@ import (
 	"testing"
 
 	. "github.com/cloudevents/sdk-go/v2/test"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
-	"knative.dev/eventing-kafka-broker/test/e2e_new/resources/configmap"
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/eventing/test/rekt/resources/trigger"
 	"knative.dev/pkg/system"
@@ -35,6 +33,9 @@ import (
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
 	"knative.dev/reconciler-test/resources/svc"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	"knative.dev/eventing-kafka-broker/test/e2e_new/resources/configmap"
 )
 
 func TestBrokerDeleteContractConfigMap(t *testing.T) {
diff --git a/test/e2e_new/broker_not_ready_test.go b/test/e2e_new/broker_not_ready_test.go
index 2e319e60a..42948656a 100644
--- a/test/e2e_new/broker_not_ready_test.go
+++ b/test/e2e_new/broker_not_ready_test.go
@@ -22,14 +22,15 @@ package e2e_new
 import (
 	"testing"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
-	brokerconfigmap "knative.dev/eventing-kafka-broker/test/e2e_new/resources/configmap/broker"
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/pkg/system"
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	brokerconfigmap "knative.dev/eventing-kafka-broker/test/e2e_new/resources/configmap/broker"
 )
 
 func TestBrokerNotReadyAfterBeingReady(t *testing.T) {
diff --git a/test/e2e_new/main_test.go b/test/e2e_new/main_test.go
index c31e03102..7d02eae8c 100644
--- a/test/e2e_new/main_test.go
+++ b/test/e2e_new/main_test.go
@@ -33,7 +33,7 @@ import (
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/reconciler-test/pkg/environment"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 )
 
 // global is the singleton instance of GlobalEnvironment. It is used to parse
diff --git a/test/e2e_new/ordered_test.go b/test/e2e_new/ordered_test.go
index e6f5e4d99..a815f001f 100644
--- a/test/e2e_new/ordered_test.go
+++ b/test/e2e_new/ordered_test.go
@@ -31,9 +31,6 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	cetest "github.com/cloudevents/sdk-go/v2/test"
 	"github.com/stretchr/testify/require"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
-	"knative.dev/eventing-kafka-broker/test/e2e_new/multiple_partition_config"
-	"knative.dev/eventing-kafka-broker/test/e2e_new/single_partition_config"
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/eventing/test/rekt/resources/trigger"
 	"knative.dev/pkg/system"
@@ -43,6 +40,10 @@ import (
 	"knative.dev/reconciler-test/pkg/knative"
 	"knative.dev/reconciler-test/resources/svc"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	"knative.dev/eventing-kafka-broker/test/e2e_new/multiple_partition_config"
+	"knative.dev/eventing-kafka-broker/test/e2e_new/single_partition_config"
+
 	. "knative.dev/reconciler-test/pkg/eventshub/assert"
 )
 
diff --git a/test/e2e_new/tracing_test.go b/test/e2e_new/tracing_test.go
index e4f2c4fe4..95b921629 100644
--- a/test/e2e_new/tracing_test.go
+++ b/test/e2e_new/tracing_test.go
@@ -35,9 +35,9 @@ import (
 	"knative.dev/reconciler-test/pkg/knative"
 	"knative.dev/reconciler-test/resources/svc"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
-
 	. "knative.dev/reconciler-test/pkg/eventshub/assert"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 )
 
 func TestTracingHeaders(t *testing.T) {
diff --git a/test/pkg/broker/broker.go b/test/pkg/broker/broker.go
index c0b6daa92..6472d9d90 100644
--- a/test/pkg/broker/broker.go
+++ b/test/pkg/broker/broker.go
@@ -28,7 +28,7 @@ import (
 	"knative.dev/eventing/test/lib/resources"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
diff --git a/test/test_images/kafka-consumer/consumer.go b/test/test_images/kafka-consumer/consumer.go
index 2eaa8c57a..b3a3e1dae 100644
--- a/test/test_images/kafka-consumer/consumer.go
+++ b/test/test_images/kafka-consumer/consumer.go
@@ -33,7 +33,7 @@ import (
 	"knative.dev/pkg/signals"
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	kafkatest "knative.dev/eventing-kafka-broker/test/pkg/kafka"
 )
 
diff --git a/test/upgrade/continual/broker.go b/test/upgrade/continual/broker.go
index 42801c4c4..7776a9b0f 100644
--- a/test/upgrade/continual/broker.go
+++ b/test/upgrade/continual/broker.go
@@ -35,7 +35,7 @@ import (
 
 	eventingkafkaupgrade "knative.dev/eventing-kafka/test/upgrade/continual"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )