Repository: knative-sandbox/eventing-kafka-broker
Pull request #42 (2020-07-16 08:37:09)

Add Trigger reconciler
## Proposed Changes

- Add Trigger reconciler

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/42#" 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 **evankanderson**
You can assign the PR to them by writing `/assign @evankanderson` in a comment when ready.

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

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/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":["evankanderson"]} -->
pierDipi: I'll do the Finalizer part in another PR.

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/42/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1283682157155323904/artifacts/line-cov.html#file0) | 90.4% | 90.4% | -0.1
[control-plane/pkg/reconciler/trigger/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/42/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1283682157155323904/artifacts/line-cov.html#file1) | 50.0% | 42.9% | -7.1
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/42/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1283682157155323904/artifacts/line-cov.html#file2) | 60.0% | 86.5% | 26.5
[control-plane/pkg/reconciler/trigger/trigger_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/42/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1283682157155323904/artifacts/line-cov.html#file3) | Do not exist | 61.1% | 

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-go-coverage | e81e675499eb464aca4ab860b09d68942ff21eac | [link](https://prow.knative.dev/view/gcs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/42/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1283682157155323904) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/42). [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/42/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1283687819574775808/artifacts/line-cov.html#file0) | 90.4% | 90.4% | -0.1
[control-plane/pkg/reconciler/trigger/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/42/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1283687819574775808/artifacts/line-cov.html#file1) | 50.0% | 62.1% | 12.1
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/42/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1283687819574775808/artifacts/line-cov.html#file2) | 60.0% | 86.5% | 26.5
[control-plane/pkg/reconciler/trigger/trigger_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/42/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1283687819574775808/artifacts/line-cov.html#file3) | Do not exist | 61.1% | 

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/42#" title="Author self-approved">pierDipi</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/42#pullrequestreview-449871701" 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/control-plane/cmd/kafka-broker-controller/main.go b/control-plane/cmd/kafka-broker-controller/main.go
index e16ae3ace..705394b35 100644
--- a/control-plane/cmd/kafka-broker-controller/main.go
+++ b/control-plane/cmd/kafka-broker-controller/main.go
@@ -54,6 +54,8 @@ func main() {
 			return broker.NewController(ctx, watcher, brokerConfigs)
 		},
 
-		trigger.NewController,
+		func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
+			return trigger.NewController(ctx, watcher, &brokerConfigs.EnvConfigs)
+		},
 	)
 }
diff --git a/control-plane/pkg/log/logger.go b/control-plane/pkg/log/logger.go
new file mode 100644
index 000000000..2a2cebcff
--- /dev/null
+++ b/control-plane/pkg/log/logger.go
@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+
+package log
+
+import (
+	"context"
+	"fmt"
+
+	"go.uber.org/zap"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing/pkg/logging"
+)
+
+func Logger(ctx context.Context, key string, object metav1.Object) *zap.Logger {
+
+	return logging.FromContext(ctx).With(zap.String(
+		key,
+		fmt.Sprintf("%s/%s", object.GetNamespace(), object.GetName()),
+	))
+}
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 003c357de..3641fad46 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -43,7 +43,7 @@ const (
 	TopicPrefix = "knative-broker-"
 
 	// signal that the broker hasn't been added to the config map yet.
-	noBroker = -1
+	NoBroker = -1
 )
 
 type Reconciler struct {
@@ -63,7 +63,7 @@ type Reconciler struct {
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
 
-	logger := logger(ctx, broker)
+	logger := log.Logger(ctx, "broker", broker)
 
 	logger.Debug("Reconciling broker")
 
@@ -100,7 +100,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, broker *eventing.Broker)
 		zap.Any(base.BrokersTriggersDataLogKey, log.BrokersMarshaller{Brokers: brokersTriggers}),
 	)
 
-	brokerIndex := findBroker(brokersTriggers, broker)
+	brokerIndex := FindBroker(brokersTriggers, broker)
 
 	// Get broker configuration.
 	brokerConfig, err := r.getBrokerConfig(topic, broker)
@@ -108,7 +108,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, broker *eventing.Broker)
 		return statusConditionManager.failedToGetBrokerConfig(err)
 	}
 	// Update brokersTriggers data with the new broker configuration
-	if brokerIndex != noBroker {
+	if brokerIndex != NoBroker {
 		brokersTriggers.Broker[brokerIndex] = brokerConfig
 
 		logger.Debug("Broker exists", zap.Int("index", brokerIndex))
@@ -141,6 +141,8 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, broker *eventing.Broker)
 		return statusConditionManager.failedToUpdateReceiverPodsAnnotation(err)
 	}
 
+	logger.Debug("Updated receiver pod annotation")
+
 	// Update volume generation annotation of dispatcher pods
 	if err := r.UpdateDispatcherPodsAnnotation(logger, brokersTriggers.VolumeGeneration); err != nil {
 		// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
@@ -156,13 +158,11 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, broker *eventing.Broker)
 		logger.Debug("Updated dispatcher pod annotation")
 	}
 
-	logger.Debug("Updated receiver pod annotation")
-
 	return statusConditionManager.reconciled()
 }
 
 func (r *Reconciler) FinalizeKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
-	logger := logger(ctx, broker)
+	logger := log.Logger(ctx, "broker", broker)
 
 	logger.Debug("Finalizing broker")
 
@@ -185,8 +185,8 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, broker *eventing.Broker)
 		zap.Any(base.BrokersTriggersDataLogKey, log.BrokersMarshaller{Brokers: brokersTriggers}),
 	)
 
-	brokerIndex := findBroker(brokersTriggers, broker)
-	if brokerIndex != noBroker {
+	brokerIndex := FindBroker(brokersTriggers, broker)
+	if brokerIndex != NoBroker {
 		deleteBroker(brokersTriggers, brokerIndex)
 
 		logger.Debug("Broker deleted", zap.Int("index", brokerIndex))
@@ -313,9 +313,9 @@ func (r *Reconciler) SetDefaultTopicDetails(topicDetail sarama.TopicDetail) {
 	r.KafkaDefaultTopicDetails = topicDetail
 }
 
-func findBroker(brokersTriggers *coreconfig.Brokers, broker *eventing.Broker) int {
+func FindBroker(brokersTriggers *coreconfig.Brokers, broker *eventing.Broker) int {
 	// Find broker in brokersTriggers.
-	brokerIndex := noBroker
+	brokerIndex := NoBroker
 	for i, b := range brokersTriggers.Broker {
 		if b.Id == string(broker.UID) {
 			brokerIndex = i
@@ -338,11 +338,3 @@ func deleteBroker(brokersTriggers *coreconfig.Brokers, index int) {
 	// truncate the array.
 	brokersTriggers.Broker = brokersTriggers.Broker[:len(brokersTriggers.Broker)-1]
 }
-
-func logger(ctx context.Context, broker *eventing.Broker) *zap.Logger {
-
-	return logging.FromContext(ctx).With(zap.String(
-		"broker",
-		fmt.Sprintf("%s/%s", broker.Namespace, broker.Name),
-	))
-}
diff --git a/control-plane/pkg/reconciler/trigger/controller.go b/control-plane/pkg/reconciler/trigger/controller.go
index d1469575c..ed5f316d0 100644
--- a/control-plane/pkg/reconciler/trigger/controller.go
+++ b/control-plane/pkg/reconciler/trigger/controller.go
@@ -22,41 +22,94 @@ import (
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/tools/cache"
+	"knative.dev/eventing/pkg/logging"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
-	"knative.dev/pkg/logging"
+	"knative.dev/pkg/resolver"
 
-	"knative.dev/eventing/pkg/apis/eventing"
-	"knative.dev/eventing/pkg/apis/eventing/v1beta1"
+	apiseventing "knative.dev/eventing/pkg/apis/eventing"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1beta1"
 	brokerinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1beta1/broker"
 	triggerinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1beta1/trigger"
 	triggerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/trigger"
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1beta1"
 
+	"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"
 )
 
-func NewController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
+const (
+	ControllerAgentName = "kafka-trigger-controller"
+
+	FinalizerName = "kafka.triggers.eventing.knative.dev"
+)
+
+func NewController(ctx context.Context, _ configmap.Watcher, configs *broker.EnvConfigs) *controller.Impl {
+
+	logger := logging.FromContext(ctx)
 
 	brokerInformer := brokerinformer.Get(ctx)
 	triggerInformer := triggerinformer.Get(ctx)
 	triggerLister := triggerInformer.Lister()
 
 	reconciler := &Reconciler{
-		logger: logging.FromContext(ctx).Desugar(),
+		Reconciler: &base.Reconciler{
+			KubeClient:                  kubeclient.Get(ctx),
+			PodLister:                   podinformer.Get(ctx).Lister(),
+			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
+			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
+			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
+			SystemNamespace:             configs.SystemNamespace,
+		},
+		BrokerLister: brokerInformer.Lister(),
+		Configs:      configs,
+		Recorder:     controller.GetEventRecorder(ctx),
 	}
 
-	impl := triggerreconciler.NewImpl(ctx, reconciler)
+	impl := triggerreconciler.NewImpl(ctx, reconciler, func(impl *controller.Impl) controller.Options {
+		return controller.Options{
+			FinalizerName:     FinalizerName,
+			AgentName:         ControllerAgentName,
+			SkipStatusUpdates: false,
+		}
+	})
+
+	reconciler.Resolver = resolver.NewURIResolver(ctx, impl.EnqueueKey)
+
+	triggerInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: filterTriggers(reconciler.BrokerLister),
+		Handler:    controller.HandleAll(impl.Enqueue),
+	})
 
 	// Filter Brokers and enqueue associated Triggers
 	brokerInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: kafka.BrokerClassFilter(),
-		Handler:    enqueueTriggers(reconciler.logger, triggerLister, impl.Enqueue),
+		Handler:    enqueueTriggers(logger, triggerLister, impl.Enqueue),
 	})
 
 	return impl
 }
 
+func filterTriggers(lister eventinglisters.BrokerLister) func(interface{}) bool {
+	return func(obj interface{}) bool {
+		trigger, ok := obj.(*eventing.Trigger)
+		if !ok {
+			return false
+		}
+
+		broker, err := lister.Brokers(trigger.Namespace).Get(trigger.Spec.Broker)
+		if err != nil {
+			return true // TODO decide what to do on NotFound/Found
+		}
+
+		value, ok := broker.GetAnnotations()[apiseventing.BrokerClassKey]
+		return ok && value == kafka.BrokerClass
+	}
+}
+
 func enqueueTriggers(
 	logger *zap.Logger,
 	lister eventinglisters.TriggerLister,
@@ -64,9 +117,9 @@ func enqueueTriggers(
 
 	return controller.HandleAll(func(obj interface{}) {
 
-		if broker, ok := obj.(*v1beta1.Broker); ok {
+		if broker, ok := obj.(*eventing.Broker); ok {
 
-			selector := labels.SelectorFromSet(map[string]string{eventing.BrokerLabelKey: broker.Name})
+			selector := labels.SelectorFromSet(map[string]string{apiseventing.BrokerLabelKey: broker.Name})
 			triggers, err := lister.Triggers(broker.Namespace).List(selector)
 			if err != nil {
 				logger.Warn("Failed to list triggers", zap.Any("broker", broker), zap.Error(err))
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index ed2f4a6a7..367556305 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -19,38 +19,170 @@ package trigger
 import (
 	"context"
 	"fmt"
+	"math"
 
 	"go.uber.org/zap"
-	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/client-go/tools/record"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1beta1"
+	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1beta1"
 	"knative.dev/pkg/reconciler"
+	"knative.dev/pkg/resolver"
+
+	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/log"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	brokerreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
 )
 
 const (
 	trigger           = "Trigger"
 	triggerReconciled = trigger + "Reconciled"
+
+	noTrigger = brokerreconciler.NoBroker
 )
 
 type Reconciler struct {
-	logger *zap.Logger
+	*base.Reconciler
+
+	BrokerLister eventinglisters.BrokerLister
+	Resolver     *resolver.URIResolver
+
+	Configs *brokerreconciler.EnvConfigs
+
+	Recorder record.EventRecorder
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, trigger *eventing.Trigger) reconciler.Event {
-	r.logger.Debug("reconciling Trigger", zap.Any("trigger", trigger))
 
-	return reconciledNormal(trigger.Namespace, trigger.Name)
+	logger := log.Logger(ctx, "trigger", trigger)
+
+	logger.Debug("Reconciling Trigger", zap.Any("trigger", trigger))
+
+	statusConditionManager := statusConditionManager{
+		Trigger: trigger,
+		Configs: r.Configs,
+	}
+
+	broker, err := r.BrokerLister.Brokers(trigger.Namespace).Get(trigger.Spec.Broker)
+	if err != nil && !apierrors.IsNotFound(err) {
+		return statusConditionManager.failedToGetBroker(err)
+	}
+
+	if apierrors.IsNotFound(err) || !broker.GetDeletionTimestamp().IsZero() {
+		// The associated broker doesn't exist anymore, so clean up Trigger resources.
+		return r.FinalizeKind(ctx, trigger)
+	}
+
+	statusConditionManager.propagateBrokerCondition(broker)
+
+	// Get data plane config map.
+	dataPlaneConfigMap, err := r.GetDataPlaneConfigMap()
+	if err != nil {
+		return statusConditionManager.failedToGetDataPlaneConfigMap(err)
+	}
+
+	logger.Debug("Got brokers and triggers config map")
+
+	// Get data plane config data.
+	dataPlaneConfig, err := r.GetDataPlaneConfigMapData(logger, dataPlaneConfigMap)
+	if err != nil || dataPlaneConfig == nil {
+		return statusConditionManager.failedToGetDataPlaneConfigFromConfigMap(err)
+	}
+
+	logger.Debug(
+		"Got brokers and triggers data from config map",
+		zap.Any(base.BrokersTriggersDataLogKey, log.BrokersMarshaller{Brokers: dataPlaneConfig}),
+	)
+
+	brokerIndex := brokerreconciler.FindBroker(dataPlaneConfig, broker)
+	if brokerIndex == brokerreconciler.NoBroker {
+		return statusConditionManager.brokerNotFoundInDataPlaneConfigMap()
+	}
+	triggerIndex := findTrigger(dataPlaneConfig.Broker[brokerIndex].Triggers, trigger)
+
+	triggerConfig, err := r.GetTriggerConfig(trigger)
+	if err != nil {
+		return statusConditionManager.failedToResolveTriggerConfig(err)
+	}
+
+	statusConditionManager.subscriberResolved()
+
+	if triggerIndex == noTrigger {
+		dataPlaneConfig.Broker[brokerIndex].Triggers = append(
+			dataPlaneConfig.Broker[brokerIndex].Triggers,
+			&triggerConfig,
+		)
+	} else {
+		dataPlaneConfig.Broker[brokerIndex].Triggers[triggerIndex] = &triggerConfig
+	}
+
+	// Increment volumeGeneration
+	dataPlaneConfig.VolumeGeneration = incrementVolumeGeneration(dataPlaneConfig.VolumeGeneration)
+
+	// Update the configuration map with the new dataPlaneConfig data.
+	if err := r.UpdateDataPlaneConfigMap(dataPlaneConfig, dataPlaneConfigMap); err != nil {
+		return fmt.Errorf("failed to update configuration map: %w", err)
+	}
+
+	// Update volume generation annotation of dispatcher pods
+	if err := r.UpdateDispatcherPodsAnnotation(logger, dataPlaneConfig.VolumeGeneration); err != nil {
+		// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
+		// Since the dispatcher side is the consumer side, we don't lose availability, and we can consider the Trigger
+		// ready. So, log out the error and move on to the next step.
+		logger.Warn(
+			"Failed to update dispatcher pod annotation to trigger an immediate config map refresh",
+			zap.Error(err),
+		)
+
+		statusConditionManager.failedToUpdateDispatcherPodsAnnotation(err)
+	} else {
+		logger.Debug("Updated dispatcher pod annotation")
+	}
+
+	logger.Debug("Brokers and triggers config map updated")
+
+	return statusConditionManager.reconciled()
 }
 
 func (r *Reconciler) FinalizeKind(ctx context.Context, trigger *eventing.Trigger) reconciler.Event {
-	r.logger.Debug("finalizing Trigger", zap.Any("trigger", trigger))
+
+	logger := log.Logger(ctx, "trigger", trigger)
+
+	logger.Debug("Finalizing Trigger", zap.Any("trigger", trigger))
 
 	return reconciledNormal(trigger.Namespace, trigger.Name)
 }
 
-func reconciledNormal(namespace, name string) reconciler.Event {
-	return reconciler.NewEvent(
-		corev1.EventTypeNormal,
-		triggerReconciled,
-		fmt.Sprintf(`%s reconciled: "%s/%s"`, trigger, namespace, name),
-	)
+func (r *Reconciler) GetTriggerConfig(trigger *eventing.Trigger) (coreconfig.Trigger, error) {
+
+	var attributes map[string]string
+	if trigger.Spec.Filter != nil {
+		attributes = trigger.Spec.Filter.Attributes
+	}
+
+	destination, err := r.Resolver.URIFromDestinationV1(trigger.Spec.Subscriber, trigger)
+	if err != nil {
+		return coreconfig.Trigger{}, fmt.Errorf("failed to resolve Trigger.Spec.Subscriber: %w", err)
+	}
+
+	return coreconfig.Trigger{
+		Attributes:  attributes,
+		Destination: destination.String(),
+		Id:          string(trigger.UID),
+	}, nil
+}
+
+func findTrigger(triggers []*coreconfig.Trigger, trigger *eventing.Trigger) int {
+
+	for i, t := range triggers {
+		if t.Id == string(trigger.UID) {
+			return i
+		}
+	}
+	return noTrigger
+}
+
+func incrementVolumeGeneration(generation uint64) uint64 {
+	return (generation + 1) % (math.MaxUint64 - 1)
 }
diff --git a/control-plane/pkg/reconciler/trigger/trigger_lifecycle.go b/control-plane/pkg/reconciler/trigger/trigger_lifecycle.go
new file mode 100644
index 000000000..75d31217a
--- /dev/null
+++ b/control-plane/pkg/reconciler/trigger/trigger_lifecycle.go
@@ -0,0 +1,135 @@
+/*
+ * 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.
+ */
+
+package trigger
+
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/tools/record"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1beta1"
+	"knative.dev/pkg/apis"
+	"knative.dev/pkg/reconciler"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+)
+
+type statusConditionManager struct {
+	Trigger *eventing.Trigger
+
+	Configs *broker.EnvConfigs
+
+	Recorder record.EventRecorder
+}
+
+func (m *statusConditionManager) failedToGetBroker(err error) reconciler.Event {
+
+	m.Trigger.Status.MarkBrokerFailed(
+		"Failed to get broker",
+		"%v",
+		err,
+	)
+
+	return fmt.Errorf("failed to get broker: %w", err)
+}
+
+func (m *statusConditionManager) failedToGetDataPlaneConfigMap(err error) reconciler.Event {
+
+	reason := fmt.Sprintf("Failed to get data plane config map %s", m.Configs.DataPlaneConfigMapAsString())
+	m.Trigger.Status.MarkDependencyFailed(
+		reason,
+		"%v",
+		err,
+	)
+
+	return fmt.Errorf(reason+": %w", err)
+}
+
+func (m *statusConditionManager) propagateBrokerCondition(broker *eventing.Broker) {
+	m.Trigger.Status.PropagateBrokerCondition(broker.Status.GetTopLevelCondition())
+}
+
+func (m *statusConditionManager) failedToGetDataPlaneConfigFromConfigMap(err error) reconciler.Event {
+
+	m.Trigger.Status.MarkDependencyFailed(
+		"Failed to get data plane config from config map",
+		"%v",
+		err,
+	)
+
+	return fmt.Errorf("failed to get data plane config from config map: %w", err)
+}
+
+func (m *statusConditionManager) brokerNotFoundInDataPlaneConfigMap() reconciler.Event {
+
+	m.Trigger.Status.MarkBrokerFailed(
+		"Broker not found in data plane map",
+		"config map: %s",
+		m.Configs.DataPlaneConfigMapAsString(),
+	)
+
+	return fmt.Errorf("broker not found in data plane config map %s", m.Configs.DataPlaneConfigMapAsString())
+}
+
+func (m *statusConditionManager) reconciled() reconciler.Event {
+
+	m.Trigger.Status.MarkDependencySucceeded()
+
+	// TODO we don't have a subscription, consider register custom condition set for Triggers
+	m.Trigger.Status.PropagateSubscriptionCondition(&apis.Condition{
+		Status: corev1.ConditionTrue,
+	})
+
+	return reconciledNormal(m.Trigger.Namespace, m.Trigger.Name)
+}
+
+func (m *statusConditionManager) failedToResolveTriggerConfig(err error) reconciler.Event {
+
+	m.Trigger.Status.MarkSubscriberResolvedFailed(
+		"Failed to resolve trigger config",
+		"%v",
+		err,
+	)
+
+	return fmt.Errorf("failed to resolve trigger config: %w", err)
+}
+
+func (m *statusConditionManager) failedToUpdateDispatcherPodsAnnotation(err error) {
+
+	// We don't set status conditions for dispatcher pods updates.
+
+	// Record the event.
+	m.Recorder.Eventf(
+		m.Trigger,
+		corev1.EventTypeWarning,
+		"Failed to update dispatcher pods annotation",
+		"%v",
+		err,
+	)
+}
+
+func (m *statusConditionManager) subscriberResolved() {
+	m.Trigger.Status.MarkSubscriberResolvedSucceeded()
+}
+
+func reconciledNormal(namespace, name string) reconciler.Event {
+	return reconciler.NewEvent(
+		corev1.EventTypeNormal,
+		triggerReconciled,
+		fmt.Sprintf(`%s reconciled: "%s/%s"`, trigger, namespace, name),
+	)
+}


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 1958e6b90..bf27b8ede 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -18,7 +18,6 @@ package broker_test // different package name due to import cycles. (broker -> t
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"net/url"
 	"sync"
@@ -26,16 +25,13 @@ import (
 	"time"
 
 	"github.com/Shopify/sarama"
-	"github.com/gogo/protobuf/proto"
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	clientgotesting "k8s.io/client-go/testing"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1beta1"
-	"knative.dev/eventing/pkg/reconciler/names"
 	"knative.dev/pkg/apis"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	"knative.dev/pkg/controller"
@@ -57,14 +53,6 @@ import (
 )
 
 const (
-	brokerNamespace = "test-namespace"
-	brokerName      = "test-broker"
-
-	serviceNamespace = "test-service-namespace"
-	serviceName      = "test-service"
-
-	brokerUUID = "e7185016-5d98-4b54-84e8-3b1cd4acc6b4"
-
 	wantErrorOnCreateTopic = "wantErrorOnCreateTopic"
 	wantErrorOnDeleteTopic = "wantErrorOnDeleteTopic"
 )
@@ -77,13 +65,11 @@ var (
 	finalizerUpdatedEvent = Eventf(
 		corev1.EventTypeNormal,
 		"FinalizerUpdate",
-		fmt.Sprintf(`Updated %q finalizers`, brokerName),
+		fmt.Sprintf(`Updated %q finalizers`, BrokerName),
 	)
 
 	createTopicError = fmt.Errorf("failed to create topic")
 	deleteTopicError = fmt.Errorf("failed to delete topic")
-
-	formats = []string{base.Protobuf, base.Json}
 )
 
 func TestBrokeReconciler(t *testing.T) {
@@ -91,14 +77,14 @@ func TestBrokeReconciler(t *testing.T) {
 
 	t.Parallel()
 
-	for _, f := range formats {
+	for _, f := range Formats {
 		brokerReconciliation(t, f, *DefaultConfigs)
 	}
 }
 
 func brokerReconciliation(t *testing.T, format string, configs Configs) {
 
-	testKey := fmt.Sprintf("%s/%s", brokerNamespace, brokerName)
+	testKey := fmt.Sprintf("%s/%s", BrokerNamespace, BrokerName)
 
 	configs.DataPlaneConfigFormat = format
 
@@ -124,17 +110,17 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				Eventf(
 					corev1.EventTypeNormal,
 					Reconciled,
-					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, brokerNamespace, brokerName),
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, BrokerNamespace, BrokerName),
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&configs, &coreconfig.Brokers{
 					Broker: []*coreconfig.Broker{
 						{
-							Id:        brokerUUID,
-							Topic:     topic(),
-							Namespace: brokerNamespace,
-							Name:      brokerName,
+							Id:        BrokerUUID,
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Name:      BrokerName,
 						},
 					},
 					VolumeGeneration: 1,
@@ -181,18 +167,18 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				Eventf(
 					corev1.EventTypeNormal,
 					Reconciled,
-					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, brokerNamespace, brokerName),
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, BrokerNamespace, BrokerName),
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&configs, &coreconfig.Brokers{
 					Broker: []*coreconfig.Broker{
 						{
-							Id:             brokerUUID,
-							Topic:          topic(),
+							Id:             BrokerUUID,
+							Topic:          GetTopic(),
 							DeadLetterSink: "http://test-service.test-service-namespace.svc.cluster.local/",
-							Namespace:      brokerNamespace,
-							Name:           brokerName,
+							Namespace:      BrokerNamespace,
+							Name:           BrokerName,
 						},
 					},
 					VolumeGeneration: 2,
@@ -232,7 +218,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					corev1.EventTypeWarning,
 					"InternalError",
 					"failed to create topic: %s: %v",
-					topic(), createTopicError,
+					GetTopic(), createTopicError,
 				),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
@@ -302,17 +288,17 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				Eventf(
 					corev1.EventTypeNormal,
 					Reconciled,
-					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, brokerNamespace, brokerName),
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, BrokerNamespace, BrokerName),
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&configs, &coreconfig.Brokers{
 					Broker: []*coreconfig.Broker{
 						{
-							Id:        brokerUUID,
-							Topic:     topic(),
-							Namespace: brokerNamespace,
-							Name:      brokerName,
+							Id:        BrokerUUID,
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Name:      BrokerName,
 						},
 					},
 					VolumeGeneration: 1,
@@ -348,8 +334,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							Id:             "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topic:          "my-existing-topic-a",
 							DeadLetterSink: "http://www.my-sink.com",
-							Namespace:      brokerNamespace,
-							Name:           brokerName,
+							Namespace:      BrokerNamespace,
+							Name:           BrokerName,
 						},
 						{
 							Id:             "5384faa4-6bdf-428d-b6c2-d6f89ce1d44a",
@@ -372,7 +358,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				Eventf(
 					corev1.EventTypeNormal,
 					Reconciled,
-					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, brokerNamespace, brokerName),
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, BrokerNamespace, BrokerName),
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -382,8 +368,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							Id:             "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topic:          "my-existing-topic-a",
 							DeadLetterSink: "http://www.my-sink.com",
-							Namespace:      brokerNamespace,
-							Name:           brokerName,
+							Namespace:      BrokerNamespace,
+							Name:           BrokerName,
 						},
 						{
 							Id:             "5384faa4-6bdf-428d-b6c2-d6f89ce1d44a",
@@ -391,10 +377,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							DeadLetterSink: "http://www.my-sink.com",
 						},
 						{
-							Id:        brokerUUID,
-							Topic:     topic(),
-							Namespace: brokerNamespace,
-							Name:      brokerName,
+							Id:        BrokerUUID,
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Name:      BrokerName,
 						},
 					},
 					VolumeGeneration: 1,
@@ -444,8 +430,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							DeadLetterSink: "http://www.my-sink.com",
 						},
 						{
-							Id:             brokerUUID,
-							Topic:          topic(),
+							Id:             BrokerUUID,
+							Topic:          GetTopic(),
 							DeadLetterSink: "http://www.my-sink.com",
 						},
 					},
@@ -464,7 +450,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				Eventf(
 					corev1.EventTypeNormal,
 					Reconciled,
-					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, brokerNamespace, brokerName),
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, BrokerNamespace, BrokerName),
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -476,11 +462,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							DeadLetterSink: "http://www.my-sink.com",
 						},
 						{
-							Id:             brokerUUID,
-							Topic:          topic(),
+							Id:             BrokerUUID,
+							Topic:          GetTopic(),
 							DeadLetterSink: "http://www.my-sink.com/api",
-							Namespace:      brokerNamespace,
-							Name:           brokerName,
+							Namespace:      BrokerNamespace,
+							Name:           BrokerName,
 						},
 					},
 					VolumeGeneration: 1,
@@ -530,15 +516,15 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							Id:             "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topic:          "my-existing-topic-a",
 							DeadLetterSink: "http://www.my-sink.com",
-							Namespace:      brokerNamespace,
-							Name:           brokerName,
+							Namespace:      BrokerNamespace,
+							Name:           BrokerName,
 						},
 						{
-							Id:             brokerUUID,
-							Topic:          topic(),
+							Id:             BrokerUUID,
+							Topic:          GetTopic(),
 							DeadLetterSink: "http://www.my-sink.com",
-							Namespace:      brokerNamespace,
-							Name:           brokerName,
+							Namespace:      BrokerNamespace,
+							Name:           BrokerName,
 						},
 					},
 				}, &configs),
@@ -556,7 +542,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				Eventf(
 					corev1.EventTypeNormal,
 					Reconciled,
-					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, brokerNamespace, brokerName),
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, BrokerNamespace, BrokerName),
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -566,14 +552,14 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							Id:             "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topic:          "my-existing-topic-a",
 							DeadLetterSink: "http://www.my-sink.com",
-							Namespace:      brokerNamespace,
-							Name:           brokerName,
+							Namespace:      BrokerNamespace,
+							Name:           BrokerName,
 						},
 						{
-							Id:        brokerUUID,
-							Topic:     topic(),
-							Namespace: brokerNamespace,
-							Name:      brokerName,
+							Id:        BrokerUUID,
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Name:      BrokerName,
 						},
 					},
 					VolumeGeneration: 1,
@@ -612,14 +598,14 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							Id:             "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topic:          "my-existing-topic-a",
 							DeadLetterSink: "http://www.my-sink.com",
-							Namespace:      brokerNamespace,
-							Name:           brokerName,
+							Namespace:      BrokerNamespace,
+							Name:           BrokerName,
 						},
 						{
-							Id:        brokerUUID,
-							Topic:     topic(),
-							Namespace: brokerNamespace,
-							Name:      brokerName,
+							Id:        BrokerUUID,
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Name:      BrokerName,
 						},
 					},
 					VolumeGeneration: 1,
@@ -638,7 +624,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				Eventf(
 					corev1.EventTypeNormal,
 					Reconciled,
-					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, brokerNamespace, brokerName),
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, BrokerNamespace, BrokerName),
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -648,14 +634,14 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							Id:             "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topic:          "my-existing-topic-a",
 							DeadLetterSink: "http://www.my-sink.com",
-							Namespace:      brokerNamespace,
-							Name:           brokerName,
+							Namespace:      BrokerNamespace,
+							Name:           BrokerName,
 						},
 						{
-							Id:        brokerUUID,
-							Topic:     topic(),
-							Namespace: brokerNamespace,
-							Name:      brokerName,
+							Id:        BrokerUUID,
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Name:      BrokerName,
 						},
 					},
 					VolumeGeneration: 2,
@@ -697,14 +683,14 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							Id:             "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topic:          "my-existing-topic-a",
 							DeadLetterSink: "http://www.my-sink.com",
-							Namespace:      brokerNamespace,
-							Name:           brokerName,
+							Namespace:      BrokerNamespace,
+							Name:           BrokerName,
 						},
 						{
-							Id:        brokerUUID,
-							Topic:     topic(),
-							Namespace: brokerNamespace,
-							Name:      brokerName,
+							Id:        BrokerUUID,
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Name:      BrokerName,
 						},
 					},
 					VolumeGeneration: 1,
@@ -756,14 +742,14 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 func TestBrokerFinalizer(t *testing.T) {
 	t.Parallel()
 
-	for _, f := range formats {
+	for _, f := range Formats {
 		brokerFinalization(t, f, *DefaultConfigs)
 	}
 }
 
 func brokerFinalization(t *testing.T, format string, configs Configs) {
 
-	testKey := fmt.Sprintf("%s/%s", brokerNamespace, brokerName)
+	testKey := fmt.Sprintf("%s/%s", BrokerNamespace, BrokerName)
 
 	configs.DataPlaneConfigFormat = format
 
@@ -775,10 +761,10 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 				NewConfigMapFromBrokers(&coreconfig.Brokers{
 					Broker: []*coreconfig.Broker{
 						{
-							Id:        brokerUUID,
-							Topic:     topic(),
-							Namespace: brokerNamespace,
-							Name:      brokerName,
+							Id:        BrokerUUID,
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Name:      BrokerName,
 						},
 					},
 					VolumeGeneration: 1,
@@ -789,7 +775,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 				Eventf(
 					corev1.EventTypeNormal,
 					Reconciled,
-					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, brokerNamespace, brokerName),
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, BrokerNamespace, BrokerName),
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -808,11 +794,11 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 				NewConfigMapFromBrokers(&coreconfig.Brokers{
 					Broker: []*coreconfig.Broker{
 						{
-							Id:             brokerUUID,
-							Topic:          topic(),
+							Id:             BrokerUUID,
+							Topic:          GetTopic(),
 							DeadLetterSink: "http://test-service.test-service-namespace.svc.cluster.local/",
-							Namespace:      brokerNamespace,
-							Name:           brokerName,
+							Namespace:      BrokerNamespace,
+							Name:           BrokerName,
 						},
 					},
 					VolumeGeneration: 1,
@@ -823,7 +809,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 				Eventf(
 					corev1.EventTypeNormal,
 					Reconciled,
-					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, brokerNamespace, brokerName),
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, BrokerNamespace, BrokerName),
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -839,8 +825,8 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 				NewConfigMapFromBrokers(&coreconfig.Brokers{
 					Broker: []*coreconfig.Broker{
 						{
-							Id:             brokerUUID,
-							Topic:          topic(),
+							Id:             BrokerUUID,
+							Topic:          GetTopic(),
 							DeadLetterSink: "http://test-service.test-service-namespace.svc.cluster.local/",
 						},
 					},
@@ -854,7 +840,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 					corev1.EventTypeWarning,
 					"InternalError",
 					"failed to delete topic %s: %v",
-					topic(), deleteTopicError,
+					GetTopic(), deleteTopicError,
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -917,7 +903,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 							DeadLetterSink: "http://www.my-sink.com",
 						},
 						{
-							Id:             brokerUUID,
+							Id:             BrokerUUID,
 							Topic:          "my-existing-topic-b",
 							DeadLetterSink: "http://www.my-sink.com",
 						},
@@ -930,7 +916,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 				Eventf(
 					corev1.EventTypeNormal,
 					Reconciled,
-					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, brokerNamespace, brokerName),
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, BrokerNamespace, BrokerName),
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -958,7 +944,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 							DeadLetterSink: "http://www.my-sink.com",
 						},
 						{
-							Id:             brokerUUID,
+							Id:             BrokerUUID,
 							Topic:          "my-existing-topic-b",
 							DeadLetterSink: "http://www.my-sink.com",
 						},
@@ -971,7 +957,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 				Eventf(
 					corev1.EventTypeNormal,
 					Reconciled,
-					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, brokerNamespace, brokerName),
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, BrokerNamespace, BrokerName),
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -1010,7 +996,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 				Eventf(
 					corev1.EventTypeNormal,
 					Reconciled,
-					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, brokerNamespace, brokerName),
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, Broker, BrokerNamespace, BrokerName),
 				),
 			},
 		},
@@ -1045,7 +1031,7 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 		}
 
 		clusterAdmin := &MockKafkaClusterAdmin{
-			ExpectedTopicName:   fmt.Sprintf("%s%s-%s", TopicPrefix, brokerNamespace, brokerName),
+			ExpectedTopicName:   fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName),
 			ExpectedTopicDetail: defaultTopicDetail,
 			ErrorOnCreateTopic:  onCreateTopicError,
 			ErrorOnDeleteTopic:  onDeleteTopicError,
@@ -1132,161 +1118,13 @@ func TestConfigMapUpdate(t *testing.T) {
 	assert.NotNil(t, reconciler.KafkaClusterAdmin)
 }
 
-func topic() string {
-	return fmt.Sprintf("%s%s-%s", TopicPrefix, brokerNamespace, brokerName)
-}
-
-func NewDispatcherPod(namespace string, annotations map[string]string) runtime.Object {
-	return &corev1.Pod{
-		TypeMeta: metav1.TypeMeta{
-			Kind:       "Pod",
-			APIVersion: "v1",
-		},
-		ObjectMeta: metav1.ObjectMeta{
-			Name:        "kafka-broker-dispatcher",
-			Namespace:   namespace,
-			Annotations: annotations,
-			Labels: map[string]string{
-				"app": base.DispatcherLabel,
-			},
-		},
-	}
-}
-
-func NewReceiverPod(namespace string, annotations map[string]string) runtime.Object {
-	return &corev1.Pod{
-		TypeMeta: metav1.TypeMeta{
-			Kind:       "Pod",
-			APIVersion: "v1",
-		},
-		ObjectMeta: metav1.ObjectMeta{
-			Name:        "kafka-broker-receiver",
-			Namespace:   namespace,
-			Annotations: annotations,
-			Labels: map[string]string{
-				"app": base.ReceiverLabel,
-			},
-		},
-	}
-}
-
-func DispatcherPodUpdate(namespace string, annotations map[string]string) clientgotesting.UpdateActionImpl {
-	return clientgotesting.NewUpdateAction(
-		schema.GroupVersionResource{
-			Group:    "*",
-			Version:  "v1",
-			Resource: "Pod",
-		},
-		namespace,
-		NewDispatcherPod(namespace, annotations),
-	)
-}
-
-func ReceiverPodUpdate(namespace string, annotations map[string]string) clientgotesting.UpdateActionImpl {
-	return clientgotesting.NewUpdateAction(
-		schema.GroupVersionResource{
-			Group:    "*",
-			Version:  "v1",
-			Resource: "Pod",
-		},
-		namespace,
-		NewReceiverPod(namespace, annotations),
-	)
-}
-
-func NewService() *corev1.Service {
-	return reconcilertesting.NewService(
-		serviceName,
-		serviceNamespace,
-		func(service *corev1.Service) {
-			service.APIVersion = "v1"
-			service.Kind = "Service"
-		},
-	)
-}
-
-func NewConfigMap(configs *Configs, data []byte) runtime.Object {
-	return reconcilertesting.NewConfigMap(
-		configs.DataPlaneConfigMapName,
-		configs.DataPlaneConfigMapNamespace,
-		func(configMap *corev1.ConfigMap) {
-			if configMap.BinaryData == nil {
-				configMap.BinaryData = make(map[string][]byte, 1)
-			}
-
-			configMap.BinaryData[base.ConfigMapDataKey] = data
-		},
-	)
-}
-
-func NewConfigMapFromBrokers(brokers *coreconfig.Brokers, configs *Configs) runtime.Object {
-	var data []byte
-	var err error
-	if configs.DataPlaneConfigFormat == base.Protobuf {
-		data, err = proto.Marshal(brokers)
-	} else {
-		data, err = json.Marshal(brokers)
-	}
-	if err != nil {
-		panic(err)
-	}
-
-	return NewConfigMap(configs, data)
-}
-
-func ConfigMapUpdate(configs *Configs, brokers *coreconfig.Brokers) clientgotesting.UpdateActionImpl {
-	return clientgotesting.NewUpdateAction(
-		schema.GroupVersionResource{
-			Group:    "*",
-			Version:  "v1",
-			Resource: "ConfigMap",
-		},
-		configs.DataPlaneConfigMapNamespace,
-		NewConfigMapFromBrokers(brokers, configs),
-	)
-}
-
-// NewBroker creates a new Broker with broker class equals to kafka.BrokerClass
-func NewBroker(options ...reconcilertesting.BrokerOption) runtime.Object {
-	return reconcilertesting.NewBroker(
-		brokerName,
-		brokerNamespace,
-		append(
-			options,
-			reconcilertesting.WithBrokerClass(kafka.BrokerClass),
-			func(broker *eventing.Broker) {
-				broker.UID = brokerUUID
-			},
-		)...,
-	)
-}
-
-func NewDeletedBroker(options ...reconcilertesting.BrokerOption) runtime.Object {
-	return NewBroker(
-		append(
-			options,
-			func(broker *eventing.Broker) {
-				broker.DeletionTimestamp = &metav1.Time{Time: time.Now()}
-			},
-		)...,
-	)
-}
-
-func WithDelivery() func(*eventing.Broker) {
-	service := NewService()
-
-	return func(broker *eventing.Broker) {
-		broker.Spec.Delivery = &eventingduck.DeliverySpec{
-			DeadLetterSink: &duckv1.Destination{
-				Ref: &duckv1.KReference{
-					Kind:       service.Kind,
-					Namespace:  service.Namespace,
-					Name:       service.Name,
-					APIVersion: service.APIVersion,
-				},
-			},
-		}
-	}
+func patchFinalizers() clientgotesting.PatchActionImpl {
+	action := clientgotesting.PatchActionImpl{}
+	action.Name = BrokerName
+	action.Namespace = BrokerNamespace
+	patch := `{"metadata":{"finalizers":["` + finalizerName + `"],"resourceVersion":""}}`
+	action.Patch = []byte(patch)
+	return action
 }
 
 func getUnmarshallableError(format string) interface{} {
@@ -1295,71 +1133,3 @@ func getUnmarshallableError(format string) interface{} {
 	}
 	return "invalid character '-' after object key"
 }
-
-func ConfigMapUpdatedReady(configs *Configs) func(broker *eventing.Broker) {
-	return func(broker *eventing.Broker) {
-		broker.GetConditionSet().Manage(broker.GetStatus()).MarkTrueWithReason(
-			ConditionConfigMapUpdated,
-			fmt.Sprintf("Config map %s updated", configs.DataPlaneConfigMapAsString()),
-			"",
-		)
-	}
-}
-
-func TopicReady(broker *eventing.Broker) {
-	broker.GetConditionSet().Manage(broker.GetStatus()).MarkTrueWithReason(
-		ConditionTopicReady,
-		fmt.Sprintf("Topic %s created", Topic(broker)),
-		"",
-	)
-}
-
-func Addressable(configs *Configs) func(broker *eventing.Broker) {
-
-	return func(broker *eventing.Broker) {
-
-		broker.Status.Address.URL = &apis.URL{
-			Scheme: "http",
-			Host:   names.ServiceHostName(configs.BrokerIngressName, configs.SystemNamespace),
-			Path:   fmt.Sprintf("/%s/%s", broker.Namespace, broker.Name),
-		}
-
-		broker.GetConditionSet().Manage(&broker.Status).MarkTrue(ConditionAddressable)
-	}
-}
-
-func FailedToCreateTopic(broker *eventing.Broker) {
-
-	broker.GetConditionSet().Manage(broker.GetStatus()).MarkFalse(
-		ConditionTopicReady,
-		fmt.Sprintf("Failed to create topic: %s", topic()),
-		"%v",
-		fmt.Errorf("failed to create topic"),
-	)
-
-}
-
-func FailedToGetConfigMap(configs *Configs) func(broker *eventing.Broker) {
-
-	return func(broker *eventing.Broker) {
-
-		broker.GetConditionSet().Manage(broker.GetStatus()).MarkFalse(
-			ConditionConfigMapUpdated,
-			fmt.Sprintf(
-				"Failed to get ConfigMap: %s",
-				configs.DataPlaneConfigMapAsString(),
-			),
-			`configmaps "knative-eventing" not found`,
-		)
-	}
-
-}
-
-func patchFinalizers() clientgotesting.PatchActionImpl {
-	action := clientgotesting.PatchActionImpl{}
-	action.Name = brokerName
-	action.Namespace = brokerNamespace
-	patch := `{"metadata":{"finalizers":["` + finalizerName + `"],"resourceVersion":""}}`
-	action.Patch = []byte(patch)
-	return action
-}
diff --git a/control-plane/pkg/reconciler/testing/objects.go b/control-plane/pkg/reconciler/testing/objects.go
new file mode 100644
index 000000000..2e600acef
--- /dev/null
+++ b/control-plane/pkg/reconciler/testing/objects.go
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+
+package testing
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/gogo/protobuf/proto"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	clientgotesting "k8s.io/client-go/testing"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1beta1"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1beta1"
+	"knative.dev/eventing/pkg/reconciler/names"
+	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1beta1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+
+	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
+	"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"
+)
+
+const (
+	BrokerNamespace = "test-namespace"
+	BrokerName      = "test-broker"
+
+	serviceNamespace = "test-service-namespace"
+	serviceName      = "test-service"
+	ServiceURL       = "http://test-service.test-service-namespace.svc.cluster.local/"
+
+	BrokerUUID  = "e7185016-5d98-4b54-84e8-3b1cd4acc6b4"
+	TriggerUUID = "e7185016-5d98-4b54-84e8-3b1cd4acc6b5"
+)
+
+var (
+	Formats = []string{base.Protobuf, base.Json}
+)
+
+func GetTopic() string {
+	return fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName)
+}
+
+func NewDispatcherPod(namespace string, annotations map[string]string) runtime.Object {
+	return &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        "kafka-broker-dispatcher",
+			Namespace:   namespace,
+			Annotations: annotations,
+			Labels: map[string]string{
+				"app": base.DispatcherLabel,
+			},
+		},
+	}
+}
+
+func NewReceiverPod(namespace string, annotations map[string]string) runtime.Object {
+	return &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        "kafka-broker-receiver",
+			Namespace:   namespace,
+			Annotations: annotations,
+			Labels: map[string]string{
+				"app": base.ReceiverLabel,
+			},
+		},
+	}
+}
+
+func DispatcherPodUpdate(namespace string, annotations map[string]string) clientgotesting.UpdateActionImpl {
+	return clientgotesting.NewUpdateAction(
+		schema.GroupVersionResource{
+			Group:    "*",
+			Version:  "v1",
+			Resource: "Pod",
+		},
+		namespace,
+		NewDispatcherPod(namespace, annotations),
+	)
+}
+
+func ReceiverPodUpdate(namespace string, annotations map[string]string) clientgotesting.UpdateActionImpl {
+	return clientgotesting.NewUpdateAction(
+		schema.GroupVersionResource{
+			Group:    "*",
+			Version:  "v1",
+			Resource: "Pod",
+		},
+		namespace,
+		NewReceiverPod(namespace, annotations),
+	)
+}
+
+func NewService() *corev1.Service {
+	return reconcilertesting.NewService(
+		serviceName,
+		serviceNamespace,
+		func(service *corev1.Service) {
+			service.APIVersion = "v1"
+			service.Kind = "Service"
+		},
+	)
+}
+
+func NewConfigMap(configs *Configs, data []byte) runtime.Object {
+	return reconcilertesting.NewConfigMap(
+		configs.DataPlaneConfigMapName,
+		configs.DataPlaneConfigMapNamespace,
+		func(configMap *corev1.ConfigMap) {
+			if configMap.BinaryData == nil {
+				configMap.BinaryData = make(map[string][]byte, 1)
+			}
+
+			configMap.BinaryData[base.ConfigMapDataKey] = data
+		},
+	)
+}
+
+func NewConfigMapFromBrokers(brokers *coreconfig.Brokers, configs *Configs) runtime.Object {
+	var data []byte
+	var err error
+	if configs.DataPlaneConfigFormat == base.Protobuf {
+		data, err = proto.Marshal(brokers)
+	} else {
+		data, err = json.Marshal(brokers)
+	}
+	if err != nil {
+		panic(err)
+	}
+
+	return NewConfigMap(configs, data)
+}
+
+func ConfigMapUpdate(configs *Configs, brokers *coreconfig.Brokers) clientgotesting.UpdateActionImpl {
+	return clientgotesting.NewUpdateAction(
+		schema.GroupVersionResource{
+			Group:    "*",
+			Version:  "v1",
+			Resource: "ConfigMap",
+		},
+		configs.DataPlaneConfigMapNamespace,
+		NewConfigMapFromBrokers(brokers, configs),
+	)
+}
+
+// NewBroker creates a new Broker with broker class equals to kafka.BrokerClass
+func NewBroker(options ...reconcilertesting.BrokerOption) runtime.Object {
+	return reconcilertesting.NewBroker(
+		BrokerName,
+		BrokerNamespace,
+		append(
+			options,
+			reconcilertesting.WithBrokerClass(kafka.BrokerClass),
+			func(broker *eventing.Broker) {
+				broker.UID = BrokerUUID
+			},
+		)...,
+	)
+}
+
+func NewDeletedBroker(options ...reconcilertesting.BrokerOption) runtime.Object {
+	return NewBroker(
+		append(
+			options,
+			func(broker *eventing.Broker) {
+				broker.DeletionTimestamp = &metav1.Time{Time: time.Now()}
+			},
+		)...,
+	)
+}
+
+func WithDelivery() func(*eventing.Broker) {
+	service := NewService()
+
+	return func(broker *eventing.Broker) {
+		broker.Spec.Delivery = &eventingduck.DeliverySpec{
+			DeadLetterSink: &duckv1.Destination{
+				Ref: &duckv1.KReference{
+					Kind:       service.Kind,
+					Namespace:  service.Namespace,
+					Name:       service.Name,
+					APIVersion: service.APIVersion,
+				},
+			},
+		}
+	}
+}
+
+func BrokerReady(broker *eventing.Broker) {
+	broker.Status.Conditions = duckv1.Conditions{
+		{
+			Type:   apis.ConditionReady,
+			Status: corev1.ConditionTrue,
+		},
+	}
+}
+
+func ConfigMapUpdatedReady(configs *Configs) func(broker *eventing.Broker) {
+	return func(broker *eventing.Broker) {
+		broker.GetConditionSet().Manage(broker.GetStatus()).MarkTrueWithReason(
+			ConditionConfigMapUpdated,
+			fmt.Sprintf("Config map %s updated", configs.DataPlaneConfigMapAsString()),
+			"",
+		)
+	}
+}
+
+func TopicReady(broker *eventing.Broker) {
+	broker.GetConditionSet().Manage(broker.GetStatus()).MarkTrueWithReason(
+		ConditionTopicReady,
+		fmt.Sprintf("Topic %s created", Topic(broker)),
+		"",
+	)
+}
+
+func Addressable(configs *Configs) func(broker *eventing.Broker) {
+
+	return func(broker *eventing.Broker) {
+
+		broker.Status.Address.URL = &apis.URL{
+			Scheme: "http",
+			Host:   names.ServiceHostName(configs.BrokerIngressName, configs.SystemNamespace),
+			Path:   fmt.Sprintf("/%s/%s", broker.Namespace, broker.Name),
+		}
+
+		broker.GetConditionSet().Manage(&broker.Status).MarkTrue(ConditionAddressable)
+	}
+}
+
+func FailedToCreateTopic(broker *eventing.Broker) {
+
+	broker.GetConditionSet().Manage(broker.GetStatus()).MarkFalse(
+		ConditionTopicReady,
+		fmt.Sprintf("Failed to create topic: %s", GetTopic()),
+		"%v",
+		fmt.Errorf("failed to create topic"),
+	)
+
+}
+
+func FailedToGetConfigMap(configs *Configs) func(broker *eventing.Broker) {
+
+	return func(broker *eventing.Broker) {
+
+		broker.GetConditionSet().Manage(broker.GetStatus()).MarkFalse(
+			ConditionConfigMapUpdated,
+			fmt.Sprintf(
+				"Failed to get ConfigMap: %s",
+				configs.DataPlaneConfigMapAsString(),
+			),
+			`configmaps "knative-eventing" not found`,
+		)
+	}
+
+}
diff --git a/control-plane/pkg/reconciler/trigger/controller_test.go b/control-plane/pkg/reconciler/trigger/controller_test.go
index 0641d3478..bf5443f71 100644
--- a/control-plane/pkg/reconciler/trigger/controller_test.go
+++ b/control-plane/pkg/reconciler/trigger/controller_test.go
@@ -19,19 +19,38 @@ package trigger
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1beta1"
+	brokerinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1beta1/broker"
 	"knative.dev/pkg/configmap"
 
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
 	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1beta1/broker/fake"
 	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1beta1/trigger/fake"
+	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+
+	brokerreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
 )
 
 func TestNewController(t *testing.T) {
 	ctx, _ := reconcilertesting.SetupFakeContext(t)
 
-	controller := NewController(ctx, configmap.NewStaticWatcher())
+	controller := NewController(ctx, configmap.NewStaticWatcher(), &brokerreconciler.EnvConfigs{})
 	if controller == nil {
 		t.Error("failed to create controller: <nil>")
 	}
 }
+
+func TestFilterTriggers(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	pass := filterTriggers(brokerinformer.Get(ctx).Lister())(&eventing.Trigger{
+		Spec: eventing.TriggerSpec{
+			Broker: "not-exists",
+		},
+	})
+
+	assert.True(t, pass)
+}
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index ed8969862..7ae7e4be7 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -23,14 +23,20 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	clientgotesting "k8s.io/client-go/testing"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1beta1"
 	fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
 	triggerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1beta1/trigger"
 	"knative.dev/eventing/pkg/logging"
 	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1beta1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	"knative.dev/pkg/controller"
 	. "knative.dev/pkg/reconciler/testing"
+	"knative.dev/pkg/resolver"
 
+	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
+	"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/testing"
 )
@@ -40,10 +46,6 @@ const (
 	triggerName = "test-trigger"
 	// namespace of the trigger under test
 	triggerNamespace = "test-namespace"
-	// broker name associated with trigger under test
-	brokerName = "test-broker"
-
-	finalizerName = "triggers.eventing.knative.dev"
 )
 
 var (
@@ -54,18 +56,284 @@ var (
 	)
 )
 
-func TestTriggerReconciliation(t *testing.T) {
+func TestTriggerReconciler(t *testing.T) {
+	eventing.RegisterAlternateBrokerConditionSet(broker.ConditionSet)
+
+	t.Parallel()
+
+	for _, f := range Formats {
+		triggerReconciliation(t, f, *DefaultConfigs)
+	}
+}
+
+func triggerReconciliation(t *testing.T, format string, configs broker.Configs) {
 
-	// TODO re-enable test when Trigger reconciler is ready.
 	testKey := fmt.Sprintf("%s/%s", triggerNamespace, triggerName)
 
-	configs := *DefaultConfigs
+	configs.DataPlaneConfigFormat = format
 
 	table := TableTest{
 		{
-			Name: "Reconciled normal",
+			Name: "Reconciled normal - no existing Trigger",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewService(),
+				NewConfigMapFromBrokers(&coreconfig.Brokers{
+					Broker: []*coreconfig.Broker{
+						{
+							Id:        BrokerUUID,
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Name:      BrokerName,
+						},
+					},
+				}, &configs),
+				NewDispatcherPod(configs.SystemNamespace, nil),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeNormal,
+					triggerReconciled,
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, trigger, triggerNamespace, triggerName),
+				),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &coreconfig.Brokers{
+					Broker: []*coreconfig.Broker{
+						{
+							Id:        BrokerUUID,
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Name:      BrokerName,
+							Triggers: []*coreconfig.Trigger{
+								{
+									Destination: ServiceURL,
+									Id:          TriggerUUID,
+								},
+							},
+						},
+					},
+					VolumeGeneration: 1,
+				}),
+				DispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerSubscribed(),
+						reconcilertesting.WithTriggerDependencyReady(),
+						reconcilertesting.WithTriggerBrokerReady(),
+						reconcilertesting.WithTriggerSubscriberResolvedSucceeded(),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - with existing Triggers and Brokers",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewService(),
+				NewConfigMapFromBrokers(&coreconfig.Brokers{
+					Broker: []*coreconfig.Broker{
+						{
+							Id:        BrokerUUID + "z",
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Triggers: []*coreconfig.Trigger{
+								{
+									Destination: ServiceURL,
+									Id:          TriggerUUID,
+								},
+							},
+							Name: BrokerName,
+						},
+						{
+							Id:        BrokerUUID,
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Triggers: []*coreconfig.Trigger{
+								{
+									Attributes: map[string]string{
+										"source": "source_value",
+									},
+									Destination: ServiceURL,
+									Id:          TriggerUUID + "a",
+								},
+								{
+									Attributes: map[string]string{
+										"source": "source_value",
+									},
+									Destination: ServiceURL,
+									Id:          TriggerUUID,
+								},
+							},
+							Name: BrokerName,
+						},
+					},
+					VolumeGeneration: 2,
+				}, &configs),
+				NewDispatcherPod(configs.SystemNamespace, nil),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeNormal,
+					triggerReconciled,
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, trigger, triggerNamespace, triggerName),
+				),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &coreconfig.Brokers{
+					Broker: []*coreconfig.Broker{
+						{
+							Id:        BrokerUUID + "z",
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Triggers: []*coreconfig.Trigger{
+								{
+									Destination: ServiceURL,
+									Id:          TriggerUUID,
+								},
+							},
+							Name: BrokerName,
+						},
+						{
+							Id:        BrokerUUID,
+							Topic:     GetTopic(),
+							Namespace: BrokerNamespace,
+							Triggers: []*coreconfig.Trigger{
+								{
+									Attributes: map[string]string{
+										"source": "source_value",
+									},
+									Destination: ServiceURL,
+									Id:          TriggerUUID + "a",
+								},
+								{
+									Destination: ServiceURL,
+									Id:          TriggerUUID,
+								},
+							},
+							Name: BrokerName,
+						},
+					},
+					VolumeGeneration: 3,
+				}),
+				DispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "3",
+				}),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerSubscribed(),
+						reconcilertesting.WithTriggerDependencyReady(),
+						reconcilertesting.WithTriggerBrokerReady(),
+						reconcilertesting.WithTriggerSubscriberResolvedSucceeded(),
+					),
+				},
+			},
+		},
+		{
+			Name: "Data plane config map not found",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewService(),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			Key:     testKey,
+			WantErr: true,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					fmt.Sprintf(
+						`Failed to get data plane config map %s: configmaps "knative-eventing" not found`,
+						configs.DataPlaneConfigMapAsString(),
+					),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						reconcilertesting.WithTriggerDependencyFailed(
+							fmt.Sprintf("Failed to get data plane config map %s", configs.DataPlaneConfigMapAsString()),
+							`configmaps "knative-eventing" not found`,
+						),
+					),
+				},
+			},
+		},
+		{
+			Name: "Empty data plane config map",
 			Objects: []runtime.Object{
+				NewBroker(BrokerReady),
 				newTrigger(),
+				NewService(),
+				NewConfigMap(&configs, nil),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			Key:     testKey,
+			WantErr: true,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					fmt.Sprintf(
+						"broker not found in data plane config map %s",
+						configs.DataPlaneConfigMapAsString(),
+					),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerFailed(
+							"Broker not found in data plane map",
+							fmt.Sprintf("config map: %s", configs.DataPlaneConfigMapAsString()),
+						),
+					),
+				},
+			},
+		},
+		{
+			Name: "Broker not found", // TODO add real assertions when finalizer is ready
+			Objects: []runtime.Object{
+				newTrigger(),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -76,25 +344,66 @@ func TestTriggerReconciliation(t *testing.T) {
 					fmt.Sprintf(`%s reconciled: "%s/%s"`, trigger, triggerNamespace, triggerName),
 				),
 			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+					),
+				},
+			},
+		},
+		{
+			Name: "Broker deleted", // TODO add real assertions when finalizer is ready
+			Objects: []runtime.Object{
+				newTrigger(),
+				NewDeletedBroker(),
+			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
 			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeNormal,
+					triggerReconciled,
+					fmt.Sprintf(`%s reconciled: "%s/%s"`, trigger, triggerNamespace, triggerName),
+				),
+			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
-					Object: newTrigger(reconcilertesting.WithInitTriggerConditions),
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+					),
 				},
 			},
 		},
 	}
 
+	for i := range table {
+		table[i].Name = table[i].Name + " - " + format
+	}
+
 	table.Test(t, NewFactory(&configs, func(ctx context.Context, listers *Listers, configs *broker.Configs, row *TableRow) controller.Reconciler {
 
 		logger := logging.FromContext(ctx)
 
 		reconciler := &Reconciler{
-			logger: logger,
+			Reconciler: &base.Reconciler{
+				KubeClient:                  kubeclient.Get(ctx),
+				PodLister:                   listers.GetPodLister(),
+				DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
+				DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
+				DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
+				SystemNamespace:             configs.SystemNamespace,
+			},
+			BrokerLister: listers.GetBrokerLister(),
+			Configs:      &configs.EnvConfigs,
+			Recorder:     controller.GetEventRecorder(ctx),
 		}
 
+		reconciler.Resolver = resolver.NewURIResolver(ctx, func(name types.NamespacedName) {})
+
 		return triggerreconciler.NewReconciler(
 			ctx,
 			logger.Sugar(),
@@ -102,6 +411,11 @@ func TestTriggerReconciliation(t *testing.T) {
 			listers.GetTriggerLister(),
 			controller.GetEventRecorder(ctx),
 			reconciler,
+			controller.Options{
+				FinalizerName:     FinalizerName,
+				AgentName:         ControllerAgentName,
+				SkipStatusUpdates: false,
+			},
 		)
 	}))
 }
@@ -110,8 +424,14 @@ func newTrigger(options ...reconcilertesting.TriggerOption) runtime.Object {
 	return reconcilertesting.NewTrigger(
 		triggerName,
 		triggerNamespace,
-		brokerName,
-		options...,
+		BrokerName,
+		append(
+			options,
+			reconcilertesting.WithTriggerSubscriberURI(ServiceURL),
+			func(t *eventing.Trigger) {
+				t.UID = TriggerUUID
+			},
+		)...,
 	)
 }
 
@@ -119,7 +439,7 @@ func patchFinalizers() clientgotesting.PatchActionImpl {
 	action := clientgotesting.PatchActionImpl{}
 	action.Name = triggerName
 	action.Namespace = triggerNamespace
-	patch := `{"metadata":{"finalizers":["` + finalizerName + `"],"resourceVersion":""}}`
+	patch := `{"metadata":{"finalizers":["` + FinalizerName + `"],"resourceVersion":""}}`
 	action.Patch = []byte(patch)
 	return action
 }