Repository: knative-sandbox/eventing-rabbitmq (go)

Issue:
Deleting trigger DLQ spec should clean up resources
Repro:
- Create a broker/trigger with a DLQ on the trigger spec
- Remove the DLQ from the trigger spec
- The binding between the exchange and the queue should be deleted
- The DLQ queue should be deleted
- The dispatcher on the queue should be deleted

Pull request description:
Attempts to remove any leftover DLQ resources that are no longer needed
# Changes
- 🐛  Passively tries to remove any leftover resources. Will ignore errors as it won't affect functionality.

/kind bug

Fixes #465 

```release-note
- Removes leftover RMQ resources if a Trigger's delivery spec is removed
```



Patch:
diff --git a/pkg/rabbit/service.go b/pkg/rabbit/service.go
index dcb29f87..b491bc25 100644
--- a/pkg/rabbit/service.go
+++ b/pkg/rabbit/service.go
@@ -18,6 +18,7 @@ package rabbit
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/url"
 	"strings"
@@ -189,6 +190,54 @@ func (r *Rabbit) ReconcileBrokerDLXPolicy(ctx context.Context, args *QueueArgs)
 	}, nil
 }
 
+type DeleteResourceArgs struct {
+	Kind      interface{}
+	Name      string
+	Namespace string
+	Owner     metav1.Object
+}
+
+func (r *Rabbit) DeleteResource(ctx context.Context, args *DeleteResourceArgs) error {
+	var o metav1.Object
+	var err error
+
+	switch args.Kind.(type) {
+	case rabbitv1beta1.Queue:
+		o, err = r.RabbitmqV1beta1().Queues(args.Namespace).Get(ctx, args.Name, metav1.GetOptions{})
+	case rabbitv1beta1.Exchange:
+		o, err = r.RabbitmqV1beta1().Exchanges(args.Namespace).Get(ctx, args.Name, metav1.GetOptions{})
+	case rabbitv1beta1.Binding:
+		o, err = r.RabbitmqV1beta1().Bindings(args.Namespace).Get(ctx, args.Name, metav1.GetOptions{})
+	case rabbitv1beta1.Policy:
+		o, err = r.RabbitmqV1beta1().Policies(args.Namespace).Get(ctx, args.Name, metav1.GetOptions{})
+	default:
+		return errors.New("unsupported type")
+	}
+
+	if apierrs.IsNotFound(err) {
+		return nil
+	} else if err != nil {
+		return err
+	}
+
+	if !metav1.IsControlledBy(o, args.Owner) {
+		return fmt.Errorf("%s not owned by object: %v", o.GetResourceVersion(), args.Owner)
+	}
+
+	switch args.Kind.(type) {
+	case rabbitv1beta1.Queue:
+		return r.RabbitmqV1beta1().Queues(args.Namespace).Delete(ctx, args.Name, metav1.DeleteOptions{})
+	case rabbitv1beta1.Exchange:
+		return r.RabbitmqV1beta1().Exchanges(args.Namespace).Delete(ctx, args.Name, metav1.DeleteOptions{})
+	case rabbitv1beta1.Binding:
+		return r.RabbitmqV1beta1().Bindings(args.Namespace).Delete(ctx, args.Name, metav1.DeleteOptions{})
+	case rabbitv1beta1.Policy:
+		return r.RabbitmqV1beta1().Policies(args.Namespace).Delete(ctx, args.Name, metav1.DeleteOptions{})
+	}
+
+	return nil
+}
+
 func isReady(conditions []v1beta1.Condition) bool {
 	numConditions := len(conditions)
 	// If there are no conditions at all, the resource probably hasn't been reconciled yet => not ready
diff --git a/pkg/rabbit/types.go b/pkg/rabbit/types.go
index 5b0d6553..d27e34ee 100644
--- a/pkg/rabbit/types.go
+++ b/pkg/rabbit/types.go
@@ -35,4 +35,5 @@ type Service interface {
 	ReconcileBinding(context.Context, *BindingArgs) (Result, error)
 	ReconcileBrokerDLXPolicy(context.Context, *QueueArgs) (Result, error)
 	ReconcileDLQPolicy(context.Context, *QueueArgs) (Result, error)
+	DeleteResource(ctx context.Context, args *DeleteResourceArgs) error
 }
diff --git a/pkg/reconciler/trigger/resources/dispatcher.go b/pkg/reconciler/trigger/resources/dispatcher.go
index dfd96ea7..aa67440e 100644
--- a/pkg/reconciler/trigger/resources/dispatcher.go
+++ b/pkg/reconciler/trigger/resources/dispatcher.go
@@ -60,12 +60,8 @@ type DispatcherArgs struct {
 // MakeDispatcherDeployment creates the in-memory representation of the Broker's Dispatcher Deployment.
 func MakeDispatcherDeployment(args *DispatcherArgs) *appsv1.Deployment {
 	one := int32(1)
-	var name string
-	if args.DLX {
-		name = fmt.Sprintf("%s-dlx-dispatcher", args.Trigger.Name)
-	} else {
-		name = fmt.Sprintf("%s-dispatcher", args.Trigger.Name)
-	}
+	name := DispatcherName(args.Trigger.Name, args.DLX)
+
 	dispatcher := corev1.Container{
 		Name:  dispatcherContainerName,
 		Image: args.Image,
@@ -206,3 +202,11 @@ func DispatcherLabels(brokerName string) map[string]string {
 		"eventing.knative.dev/brokerRole": "dispatcher",
 	}
 }
+
+func DispatcherName(triggerName string, dlq bool) string {
+	if dlq {
+		return fmt.Sprintf("%s-dlx-dispatcher", triggerName)
+	} else {
+		return fmt.Sprintf("%s-dispatcher", triggerName)
+	}
+}
diff --git a/pkg/reconciler/trigger/trigger.go b/pkg/reconciler/trigger/trigger.go
index 24ff8986..6682d553 100644
--- a/pkg/reconciler/trigger/trigger.go
+++ b/pkg/reconciler/trigger/trigger.go
@@ -32,6 +32,7 @@ import (
 	"knative.dev/eventing-rabbitmq/pkg/rabbit"
 	naming "knative.dev/eventing-rabbitmq/pkg/rabbitmqnaming"
 	"knative.dev/eventing-rabbitmq/pkg/reconciler/trigger/resources"
+	rabbitv1beta1 "knative.dev/eventing-rabbitmq/third_party/pkg/apis/rabbitmq.com/v1beta1"
 	rabbitclientset "knative.dev/eventing-rabbitmq/third_party/pkg/client/clientset/versioned"
 	rabbitlisters "knative.dev/eventing-rabbitmq/third_party/pkg/client/listers/rabbitmq.com/v1beta1"
 	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
@@ -203,6 +204,8 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, t *eventingv1.Trigger) p
 			return err
 		}
 	} else {
+		// Clean up any leftover resources from a previously configured DeadLetterSink
+		r.deleteDLQResources(ctx, t)
 		// There's no Delivery spec, so just mark is as there's no DeadLetterSink Configured for it.
 		t.Status.MarkDeadLetterSinkNotConfigured()
 	}
@@ -242,7 +245,6 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, t *eventingv1.Trigger) p
 			t.Status.MarkDependencyFailed("DLQPolicyFailure", "DLQ Polcy %q is not ready", dlqPolicy.Name)
 			return nil
 		}
-
 	}
 
 	binding, err := r.reconcileBinding(ctx, broker, t)
@@ -324,6 +326,52 @@ func (r *Reconciler) reconcileDeployment(ctx context.Context, d *v1.Deployment)
 	return current, nil
 }
 
+func (r *Reconciler) deleteDLQResources(ctx context.Context, t *eventingv1.Trigger) {
+	_ = r.rabbit.DeleteResource(ctx, &rabbit.DeleteResourceArgs{
+		Kind:      rabbitv1beta1.Queue{},
+		Name:      naming.CreateTriggerDeadLetterQueueName(t),
+		Namespace: t.Namespace,
+		Owner:     t,
+	})
+
+	_ = r.rabbit.DeleteResource(ctx, &rabbit.DeleteResourceArgs{
+		Kind:      rabbitv1beta1.Exchange{},
+		Name:      naming.TriggerDLXExchangeName(t),
+		Namespace: t.Namespace,
+		Owner:     t,
+	})
+
+	_ = r.rabbit.DeleteResource(ctx, &rabbit.DeleteResourceArgs{
+		Kind:      rabbitv1beta1.Binding{},
+		Name:      naming.CreateTriggerDeadLetterQueueName(t),
+		Namespace: t.Namespace,
+		Owner:     t,
+	})
+
+	_ = r.rabbit.DeleteResource(ctx, &rabbit.DeleteResourceArgs{
+		Kind:      rabbitv1beta1.Policy{},
+		Name:      naming.CreateTriggerQueueName(t),
+		Namespace: t.Namespace,
+		Owner:     t,
+	})
+
+	_ = r.deleteDeployment(ctx, t.Namespace, resources.DispatcherName(t.Name, true), t)
+}
+
+func (r *Reconciler) deleteDeployment(ctx context.Context, namespace, name string, t *eventingv1.Trigger) error {
+	d, err := r.deploymentLister.Deployments(namespace).Get(name)
+	if apierrs.IsNotFound(err) {
+		return nil
+	} else if err != nil {
+		return err
+	}
+	if !metav1.IsControlledBy(d, t) {
+		return fmt.Errorf("deployment not owned by object: %v", t)
+	}
+
+	return r.kubeClientSet.AppsV1().Deployments(namespace).Delete(ctx, name, metav1.DeleteOptions{})
+}
+
 // reconcileDispatcherDeployment reconciles Trigger's dispatcher deployment.
 func (r *Reconciler) reconcileDispatcherDeployment(ctx context.Context, t *eventingv1.Trigger, sub *apis.URL, delivery *eventingduckv1.DeliverySpec, dlq bool) (*v1.Deployment, error) {
 	rabbitmqSecret, err := r.getRabbitmqSecret(ctx, t)


Test patch:
diff --git a/pkg/reconciler/trigger/trigger_test.go b/pkg/reconciler/trigger/trigger_test.go
index 772689c0..a4ce3fe0 100644
--- a/pkg/reconciler/trigger/trigger_test.go
+++ b/pkg/reconciler/trigger/trigger_test.go
@@ -82,8 +82,13 @@ const (
 	rabbitMQBrokerName       = "rabbitbrokerhere"
 	rabbitMQBrokerConfigName = "rabbitbrokerconfig"
 
-	triggerName = "test-trigger"
-	triggerUID  = "test-trigger-uid"
+	triggerName       = "test-trigger"
+	triggerUID        = "test-trigger-uid"
+	triggerDLQName    = "t.test-namespace.test-trigger.dlq.test-trigger-uid"
+	dlxExchangeName   = "t.test-namespace.test-trigger.dlx.test-trigger-uid"
+	dlqBindingName    = "t.test-namespace.test-trigger.dlq.test-trigger-uid"
+	dlqPolicyName     = "t.test-namespace.test-trigger.test-trigger-uid"
+	dlqDispatcherName = "test-trigger-dlx-dispatcher"
 
 	rabbitURL = "amqp://localhost:5672/%2f"
 	queueName = "test-namespace.test-trigger.broker-test-uid"
@@ -252,11 +257,11 @@ func TestReconcile(t *testing.T) {
 					triggerWithFilter(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(true)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(true, false)),
 				},
 				WantCreates: []runtime.Object{
-					createDispatcherDeployment(),
+					createDispatcherDeployment(false),
 				},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: triggerWithFilterReady(),
@@ -269,8 +274,8 @@ func TestReconcile(t *testing.T) {
 					triggerWithFilter(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(true)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(true, false)),
 				},
 				WantCreates: []runtime.Object{
 					createDispatcherDeploymentWithRetries(),
@@ -287,8 +292,8 @@ func TestReconcile(t *testing.T) {
 					triggerWithDeliverySpec(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(true)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(true, false)),
 				},
 				WantCreates: []runtime.Object{
 					createDispatcherDeploymentWithRetries(),
@@ -307,7 +312,7 @@ func TestReconcile(t *testing.T) {
 					createRabbitMQBrokerConfig(),
 				},
 				WantCreates: []runtime.Object{
-					createQueue(config),
+					createQueue(config, false),
 				},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: triggerWithQueueNotReady(),
@@ -329,7 +334,7 @@ func TestReconcile(t *testing.T) {
 					Eventf(corev1.EventTypeWarning, "InternalError", "inducing failure for create queues"),
 				},
 				WantCreates: []runtime.Object{
-					createQueue(config),
+					createQueue(config, false),
 				},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: triggerWithQueueCreateFailure(),
@@ -342,10 +347,10 @@ func TestReconcile(t *testing.T) {
 					triggerWithFilter(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
+					markReady(createQueue(config, false)),
 				},
 				WantCreates: []runtime.Object{
-					createBinding(true),
+					createBinding(true, false),
 				},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: triggerWithBindingNotReady(),
@@ -358,7 +363,7 @@ func TestReconcile(t *testing.T) {
 					triggerWithFilter(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
+					markReady(createQueue(config, false)),
 				},
 				WithReactors: []clientgotesting.ReactionFunc{
 					InduceFailure("create", "bindings"),
@@ -368,7 +373,7 @@ func TestReconcile(t *testing.T) {
 					Eventf(corev1.EventTypeWarning, "InternalError", "inducing failure for create bindings"),
 				},
 				WantCreates: []runtime.Object{
-					createBinding(true),
+					createBinding(true, false),
 				},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: triggerWithBindingCreateFailure(),
@@ -381,11 +386,11 @@ func TestReconcile(t *testing.T) {
 					triggerWithFilter(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(true)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(true, false)),
 				},
 				WantCreates: []runtime.Object{
-					createDispatcherDeployment(),
+					createDispatcherDeployment(false),
 				},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: triggerWithFilterReady(),
@@ -396,8 +401,8 @@ func TestReconcile(t *testing.T) {
 				Objects: []runtime.Object{
 					ReadyBroker(config),
 					makeSubscriberAddressableAsUnstructured(),
-					markReady(createQueue(config)),
-					markReady(createBinding(false)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(false, false)),
 					NewTrigger(triggerName, testNS, brokerName,
 						WithTriggerUID(triggerUID),
 						WithTriggerSubscriberRef(subscriberGVK, subscriberName, testNS)),
@@ -405,7 +410,7 @@ func TestReconcile(t *testing.T) {
 					createRabbitMQBrokerConfig(),
 				},
 				WantCreates: []runtime.Object{
-					createDispatcherDeployment(),
+					createDispatcherDeployment(false),
 				},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: NewTrigger(triggerName, testNS, brokerName,
@@ -424,8 +429,8 @@ func TestReconcile(t *testing.T) {
 				Objects: []runtime.Object{
 					ReadyBroker(config),
 					makeSubscriberNotAddressableAsUnstructured(),
-					markReady(createQueue(config)),
-					markReady(createBinding(false)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(false, false)),
 					NewTrigger(triggerName, testNS, brokerName,
 						WithTriggerUID(triggerUID),
 						WithTriggerSubscriberRef(subscriberGVK, subscriberName, testNS)),
@@ -456,8 +461,8 @@ func TestReconcile(t *testing.T) {
 						WithTriggerSubscriberURI(subscriberURI)),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(false)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(false, false)),
 				},
 				WithReactors: []clientgotesting.ReactionFunc{
 					InduceFailure("create", "deployments"),
@@ -467,7 +472,7 @@ func TestReconcile(t *testing.T) {
 				},
 				WantErr: true,
 				WantCreates: []runtime.Object{
-					createDispatcherDeployment(),
+					createDispatcherDeployment(false),
 				},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: NewTrigger(triggerName, testNS, brokerName,
@@ -492,8 +497,8 @@ func TestReconcile(t *testing.T) {
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
 					createDifferentDispatcherDeployment(),
-					markReady(createQueue(config)),
-					markReady(createBinding(false)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(false, false)),
 				},
 				WithReactors: []clientgotesting.ReactionFunc{
 					InduceFailure("update", "deployments"),
@@ -503,7 +508,7 @@ func TestReconcile(t *testing.T) {
 				},
 				WantErr: true,
 				WantUpdates: []clientgotesting.UpdateActionImpl{{
-					Object: createDispatcherDeployment(),
+					Object: createDispatcherDeployment(false),
 				}},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: NewTrigger(triggerName, testNS, brokerName,
@@ -527,9 +532,9 @@ func TestReconcile(t *testing.T) {
 						WithTriggerSubscriberURI(subscriberURI)),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(false)),
-					createDispatcherDeployment(),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(false, false)),
+					createDispatcherDeployment(false),
 				},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: NewTrigger(triggerName, testNS, brokerName,
@@ -551,9 +556,9 @@ func TestReconcile(t *testing.T) {
 					triggerWithFilter(),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(true)),
-					createDispatcherDeployment(),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(true, false)),
+					createDispatcherDeployment(false),
 				},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: triggerWithFilterReady(),
@@ -569,8 +574,8 @@ func TestReconcile(t *testing.T) {
 						WithInitTriggerConditions,
 						WithDependencyAnnotation(dependencyAnnotation),
 					),
-					markReady(createQueue(config)),
-					markReady(createBinding(false)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(false, false)),
 				},
 				WantEvents: []string{
 					Eventf(corev1.EventTypeWarning, "InternalError", "propagating dependency readiness: getting the dependency: pingsources.sources.knative.dev \"test-ping-source\" not found"),
@@ -601,8 +606,8 @@ func TestReconcile(t *testing.T) {
 					),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(false)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(false, false)),
 				},
 				WantErr: false,
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
@@ -630,8 +635,8 @@ func TestReconcile(t *testing.T) {
 					),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(false)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(false, false)),
 				},
 				WantErr: false,
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
@@ -659,8 +664,8 @@ func TestReconcile(t *testing.T) {
 					),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(false)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(false, false)),
 				},
 				WantErr: false,
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
@@ -686,8 +691,8 @@ func TestReconcile(t *testing.T) {
 					),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(false)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(false, false)),
 				},
 				WantErr: true,
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
@@ -717,12 +722,12 @@ func TestReconcile(t *testing.T) {
 					),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(false)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(false, false)),
 				},
 				WantErr: false,
 				WantCreates: []runtime.Object{
-					createDispatcherDeployment(),
+					createDispatcherDeployment(false),
 				},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: NewTrigger(triggerName, testNS, brokerName,
@@ -749,12 +754,12 @@ func TestReconcile(t *testing.T) {
 						WithTriggerSubscriberURI(subscriberURI)),
 					createSecret(rabbitURL),
 					createRabbitMQBrokerConfig(),
-					markReady(createQueue(config)),
-					markReady(createBinding(false)),
+					markReady(createQueue(config, false)),
+					markReady(createBinding(false, false)),
 					createDispatcherDeploymentWithParallelism(),
 				},
 				WantUpdates: []clientgotesting.UpdateActionImpl{{
-					Object: createDispatcherDeployment(),
+					Object: createDispatcherDeployment(false),
 				}},
 				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 					Object: NewTrigger(triggerName, testNS, brokerName,
@@ -769,6 +774,60 @@ func TestReconcile(t *testing.T) {
 						WithTriggerStatusSubscriberURI(subscriberURI)),
 				}},
 			},
+			{
+				Name: fmt.Sprintf("%s: DLQ resources cleaned up when delivery spec removed", name),
+				Key:  testKey,
+				Objects: []runtime.Object{
+					ReadyBroker(config),
+					triggerWithFilter(),
+					createSecret(rabbitURL),
+					createRabbitMQBrokerConfig(),
+					markReady(createQueue(config, true)),
+					markReady(createBinding(true, false)),
+					markReady(createQueue(config, false)),
+					markReady(createExchange()),
+					markReady(createBinding(true, true)),
+					createDispatcherDeployment(true),
+					markReady(createPolicy()),
+				},
+				WantCreates: []runtime.Object{
+					createDispatcherDeployment(false),
+				},
+				WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+					Object: triggerWithFilterReady(),
+				}},
+				WantDeletes: []clientgotesting.DeleteActionImpl{{
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace: testNS,
+						Resource:  rabbitv1beta1.GroupVersion.WithResource("queues"),
+					},
+					Name: triggerDLQName,
+				}, {
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace: testNS,
+						Resource:  rabbitv1beta1.GroupVersion.WithResource("exchanges"),
+					},
+					Name: dlxExchangeName,
+				}, {
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace: testNS,
+						Resource:  rabbitv1beta1.GroupVersion.WithResource("bindings"),
+					},
+					Name: dlqBindingName,
+				}, {
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace: testNS,
+						Resource:  appsv1.SchemeGroupVersion.WithResource("deployments"),
+					},
+					Name: dlqDispatcherName,
+				}, {
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace: testNS,
+						Resource:  rabbitv1beta1.GroupVersion.WithResource("policies"),
+					},
+					Name: dlqPolicyName,
+				}},
+			},
 		}...)
 	}
 	logger := logtesting.TestLogger(t)
@@ -1036,7 +1095,7 @@ func createRabbitMQBrokerConfig() *v1alpha1.RabbitmqBrokerConfig {
 	}
 }
 
-func createDispatcherDeployment() *appsv1.Deployment {
+func createDispatcherDeployment(dlq bool) *appsv1.Deployment {
 	args := &resources.DispatcherArgs{
 		Trigger: &eventingv1.Trigger{
 			ObjectMeta: metav1.ObjectMeta{
@@ -1054,6 +1113,7 @@ func createDispatcherDeployment() *appsv1.Deployment {
 		BrokerUrlSecretKey: "brokerURL",
 		BrokerIngressURL:   brokerAddress,
 		Subscriber:         subscriberAddress,
+		DLX:                dlq,
 	}
 	return resources.MakeDispatcherDeployment(args)
 }
@@ -1159,7 +1219,66 @@ func makeSubscriberNotAddressableAsUnstructured() *unstructured.Unstructured {
 	}
 }
 
-func createQueue(kref *duckv1.KReference) *rabbitv1beta1.Queue {
+func createPolicy() *rabbitv1beta1.Policy {
+	t := triggerWithFilter()
+	labels := map[string]string{
+		"eventing.knative.dev/broker":  brokerName,
+		"eventing.knative.dev/trigger": triggerName,
+	}
+
+	return &rabbitv1beta1.Policy{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: testNS,
+			Name:      dlqPolicyName,
+			OwnerReferences: []metav1.OwnerReference{
+				*kmeta.NewControllerRef(t),
+			},
+			Labels: labels,
+		},
+		Spec: rabbitv1beta1.PolicySpec{
+			Name:     dlqPolicyName,
+			Priority: 0,
+			Pattern:  "*",
+			ApplyTo:  "queues",
+			RabbitmqClusterReference: rabbitv1beta1.RabbitmqClusterReference{
+				Name:      rabbitMQBrokerName,
+				Namespace: testNS,
+			},
+		},
+	}
+}
+
+func createExchange() *rabbitv1beta1.Exchange {
+	t := triggerWithFilter()
+
+	labels := map[string]string{
+		"eventing.knative.dev/broker":  brokerName,
+		"eventing.knative.dev/trigger": triggerName,
+	}
+	return &rabbitv1beta1.Exchange{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: testNS,
+			Name:      dlxExchangeName,
+			OwnerReferences: []metav1.OwnerReference{
+				*kmeta.NewControllerRef(t),
+			},
+			Labels: labels,
+		},
+		Spec: rabbitv1beta1.ExchangeSpec{
+			Name:       dlxExchangeName,
+			Vhost:      "/",
+			Type:       "headers",
+			Durable:    true,
+			AutoDelete: false,
+			RabbitmqClusterReference: rabbitv1beta1.RabbitmqClusterReference{
+				Name:      rabbitMQBrokerName,
+				Namespace: testNS,
+			},
+		},
+	}
+}
+
+func createQueue(kref *duckv1.KReference, dlq bool) *rabbitv1beta1.Queue {
 	queueType := v1alpha1.ClassicQueueType
 	if kref.Kind == "RabbitmqBrokerConfig" {
 		queueType = v1alpha1.QuorumQueueType
@@ -1169,7 +1288,7 @@ func createQueue(kref *duckv1.KReference) *rabbitv1beta1.Queue {
 		"eventing.knative.dev/trigger": triggerName,
 	}
 	t := triggerWithFilter()
-	return &rabbitv1beta1.Queue{
+	q := &rabbitv1beta1.Queue{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "t.test-namespace.test-trigger.test-trigger-uid",
 			Namespace: testNS,
@@ -1189,10 +1308,20 @@ func createQueue(kref *duckv1.KReference) *rabbitv1beta1.Queue {
 			Type: string(queueType),
 		},
 	}
+
+	if dlq {
+		q.Name = triggerDLQName
+		q.Spec.Name = triggerDLQName
+	}
+
+	return q
 }
 
-func createBinding(withFilter bool) *rabbitv1beta1.Binding {
+func createBinding(withFilter bool, dlq bool) *rabbitv1beta1.Binding {
 	bindingName := fmt.Sprintf("t.%s.%s.test-trigger-uid", testNS, triggerName)
+	if dlq {
+		bindingName = dlqBindingName
+	}
 	labels := map[string]string{
 		"eventing.knative.dev/broker":  brokerName,
 		"eventing.knative.dev/trigger": triggerName,