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

Issue:
There is no failover or protection when events fail to be sent from the RabbitMQ Source
**Describe the bug**
When there is an error sending the events from the Source to a Sink they just keeps getting reqeued and RabbitMQ goes crazy:

<img width="1104" alt="screenshot_2021-12-02_at_15 58 22" src="https://user-images.githubusercontent.com/4516177/144662159-f6666963-3a1f-41fb-9a4d-f5ca00bc9be8.png">

When the peak in message rate in the graph starts to go down is when we configured the Knative DNS in the Cluster.

 **Expected behavior**
When message are not been sent, that the Source stops and wait for the Sink to be available. And maybe that the source (if the sink is a Ref) checks that a resolvable Ref.

 **To Reproduce**
- Spun up a Cluster environment with knative, but do not configure the DNS
- Create a RabbitMQ cluster
- Create a Source with a sink (uri or ref)
- Create the sink (optional)
- Get to the RabbitMQ Management UI and see how the image above gets replicated

 **Knative release version**
1.0.0



Pull request:
Source backoff strategy on failed events
# Changes

<!-- 
Describe your changes here- ideally you can get that description straight from
your descriptive commit message(s)! 

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic
-->

- :gift: Now using CE http retries when message fails
- :gift: There is a backoff strategy when retrying
- 🧹 Not reqeueing when prefetchCount is 1 (FIFO)

<!--
In addition, categorize the changes you're making using the "/kind" Prow command, example:

/kind performance

Supported kinds are: api-change, bug, cleanup, deprecation, removal, documentation, enhancement, performance

-->
/kind performance

<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #537 #504

<!-- Please include the 'why' behind your changes if no issue exists -->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->
```release-note
Now the RabbitMQ Source, when failing to send an event, does not get stuck in a cycle requeueing messages as they come. It follows a retries based backoff strategy (and does not reqeue them if the prefetchCount is 1)
```


Patch:
diff --git a/pkg/adapter/adapter.go b/pkg/adapter/adapter.go
index 749352c5..2b23b327 100644
--- a/pkg/adapter/adapter.go
+++ b/pkg/adapter/adapter.go
@@ -22,6 +22,7 @@ import (
 	nethttp "net/http"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/NeowayLabs/wabbit"
 	"github.com/NeowayLabs/wabbit/amqp"
@@ -34,6 +35,7 @@ import (
 	"go.uber.org/zap"
 
 	"knative.dev/eventing/pkg/adapter/v2"
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/kncloudevents"
 	"knative.dev/eventing/pkg/metrics/source"
 	"knative.dev/pkg/logging"
@@ -67,12 +69,15 @@ type QueueConfig struct {
 type adapterConfig struct {
 	adapter.EnvConfig
 
-	Brokers        string `envconfig:"RABBITMQ_BROKERS" required:"true"`
-	Topic          string `envconfig:"RABBITMQ_TOPIC" required:"true"`
-	User           string `envconfig:"RABBITMQ_USER" required:"false"`
-	Password       string `envconfig:"RABBITMQ_PASSWORD" required:"false"`
-	Vhost          string `envconfig:"RABBITMQ_VHOST" required:"false"`
-	Predeclared    bool   `envconfig:"RABBITMQ_PREDECLARED" required:"false"`
+	Brokers        string        `envconfig:"RABBITMQ_BROKERS" required:"true"`
+	Topic          string        `envconfig:"RABBITMQ_TOPIC" required:"true"`
+	User           string        `envconfig:"RABBITMQ_USER" required:"false"`
+	Password       string        `envconfig:"RABBITMQ_PASSWORD" required:"false"`
+	Vhost          string        `envconfig:"RABBITMQ_VHOST" required:"false"`
+	Predeclared    bool          `envconfig:"RABBITMQ_PREDECLARED" required:"false"`
+	Retry          int           `envconfig:"HTTP_SENDER_RETRY" required:"false"`
+	BackoffPolicy  string        `envconfig:"HTTP_SENDER_BACKOFF_POLICY" required:"false"`
+	BackoffDelay   time.Duration `envconfig:"HTTP_SENDER_BACKOFF_DELAY" default:"50ms" required:"false"`
 	ChannelConfig  ChannelConfig
 	ExchangeConfig ExchangeConfig
 	QueueConfig    QueueConfig
@@ -96,6 +101,11 @@ var _ adapter.MessageAdapterConstructor = NewAdapter
 func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, httpMessageSender *kncloudevents.HTTPMessageSender, reporter source.StatsReporter) adapter.MessageAdapter {
 	logger := logging.FromContext(ctx).Desugar()
 	config := processed.(*adapterConfig)
+	if config.BackoffPolicy == "" {
+		config.BackoffPolicy = "exponential"
+	} else if config.BackoffPolicy != "linear" && config.BackoffPolicy != "exponential" {
+		logger.Sugar().Fatalf("Invalid BACKOFF_POLICY specified: must be %q or %q", v1.BackoffPolicyExponential, v1.BackoffPolicyLinear)
+	}
 
 	return &Adapter{
 		config:            config,
@@ -329,7 +339,7 @@ func (a *Adapter) processMessages(wg *sync.WaitGroup, queue <-chan wabbit.Delive
 			}
 		} else {
 			a.logger.Error("Sending event to sink failed: ", zap.Error(err))
-			err = msg.Nack(false, true)
+			err = msg.Nack(false, false)
 			if err != nil {
 				a.logger.Error("Sending Nack failed with Delivery Tag")
 			}
@@ -344,7 +354,7 @@ func (a *Adapter) postMessage(msg wabbit.Delivery) error {
 		return err
 	}
 
-	msgBinding := NewMessageFromDelivery(msg)
+	var msgBinding binding.Message = NewMessageFromDelivery(msg)
 
 	defer func() {
 		err := msgBinding.Finish(nil)
@@ -354,9 +364,7 @@ func (a *Adapter) postMessage(msg wabbit.Delivery) error {
 	}()
 
 	// if the msg is a cloudevent send it as it is to http
-	if msgBinding.ReadEncoding() != binding.EncodingUnknown {
-		err = http.WriteRequest(a.context, msgBinding, req)
-	} else {
+	if msgBinding.ReadEncoding() == binding.EncodingUnknown {
 		// if the rabbitmq msg is not a cloudevent transform it into one
 		event := cloudevents.NewEvent()
 		err = convertToCloudEvent(&event, msg, a)
@@ -365,15 +373,26 @@ func (a *Adapter) postMessage(msg wabbit.Delivery) error {
 			return err
 		}
 
-		err = http.WriteRequest(a.context, binding.ToMessage(&event), req)
+		msgBinding = binding.ToMessage(&event)
 	}
 
+	err = http.WriteRequest(a.context, msgBinding, req)
 	if err != nil {
 		a.logger.Error(fmt.Sprintf("Error writting event to http, encoding: %s", msgBinding.ReadEncoding()), zap.Error(err))
 		return err
 	}
 
-	res, err := a.httpMessageSender.Send(req)
+	backoffDelay := a.config.BackoffDelay.String()
+	backoffPolicy := (v1.BackoffPolicyType)(a.config.BackoffPolicy)
+	res, err := a.httpMessageSender.SendWithRetries(req, &kncloudevents.RetryConfig{
+		RetryMax: a.config.Retry,
+		CheckRetry: func(ctx context.Context, resp *nethttp.Response, err error) (bool, error) {
+			return kncloudevents.SelectiveRetry(ctx, resp, nil)
+		},
+		BackoffDelay:  &backoffDelay,
+		BackoffPolicy: &backoffPolicy,
+	})
+
 	if err != nil {
 		a.logger.Error("Error while sending the message", zap.Error(err))
 		return err
diff --git a/pkg/apis/sources/v1alpha1/rabbitmq_types.go b/pkg/apis/sources/v1alpha1/rabbitmq_types.go
index 3d49a06b..b5f67a6a 100644
--- a/pkg/apis/sources/v1alpha1/rabbitmq_types.go
+++ b/pkg/apis/sources/v1alpha1/rabbitmq_types.go
@@ -23,6 +23,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
@@ -128,6 +129,18 @@ type RabbitmqSourceSpec struct {
 	// RabbitMQ cluster
 	// +optional
 	Predeclared bool `json:"predeclared,omitempty"`
+	// Retry is the minimum number of retries the sender should attempt when
+	// sending an event before moving it to the dead letter sink.
+	// +optional
+	Retry *int32 `json:"retry,omitempty"`
+	// BackoffPolicy is the retry backoff policy (linear, exponential).
+	// +optional
+	BackoffPolicy *eventingduckv1.BackoffPolicyType `json:"backoffPolicy,omitempty"`
+	// BackoffDelay is the delay before retrying in time.Duration format.
+	// For linear policy, backoff delay is backoffDelay*<numberOfRetries>.
+	// For exponential policy, backoff delay is backoffDelay*2^<numberOfRetries>.
+	// +optional
+	BackoffDelay *string `json:"backoffDelay,omitempty"`
 	// ChannelConfig config for rabbitmq exchange
 	// +optional
 	ChannelConfig RabbitmqChannelConfigSpec `json:"channelConfig,omitempty"`
diff --git a/pkg/apis/sources/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/sources/v1alpha1/zz_generated.deepcopy.go
index 1b2d5d64..09714b61 100644
--- a/pkg/apis/sources/v1alpha1/zz_generated.deepcopy.go
+++ b/pkg/apis/sources/v1alpha1/zz_generated.deepcopy.go
@@ -23,7 +23,8 @@ package v1alpha1
 import (
 	corev1 "k8s.io/api/core/v1"
 	runtime "k8s.io/apimachinery/pkg/runtime"
-	v1 "knative.dev/pkg/apis/duck/v1"
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
@@ -145,12 +146,27 @@ func (in *RabbitmqSourceSpec) DeepCopyInto(out *RabbitmqSourceSpec) {
 	*out = *in
 	in.User.DeepCopyInto(&out.User)
 	in.Password.DeepCopyInto(&out.Password)
+	if in.Retry != nil {
+		in, out := &in.Retry, &out.Retry
+		*out = new(int32)
+		**out = **in
+	}
+	if in.BackoffPolicy != nil {
+		in, out := &in.BackoffPolicy, &out.BackoffPolicy
+		*out = new(v1.BackoffPolicyType)
+		**out = **in
+	}
+	if in.BackoffDelay != nil {
+		in, out := &in.BackoffDelay, &out.BackoffDelay
+		*out = new(string)
+		**out = **in
+	}
 	in.ChannelConfig.DeepCopyInto(&out.ChannelConfig)
 	out.ExchangeConfig = in.ExchangeConfig
 	out.QueueConfig = in.QueueConfig
 	if in.Sink != nil {
 		in, out := &in.Sink, &out.Sink
-		*out = new(v1.Destination)
+		*out = new(duckv1.Destination)
 		(*in).DeepCopyInto(*out)
 	}
 	return
diff --git a/pkg/reconciler/source/resources/receive_adapter.go b/pkg/reconciler/source/resources/receive_adapter.go
index dfec3b85..5c0a69f8 100644
--- a/pkg/reconciler/source/resources/receive_adapter.go
+++ b/pkg/reconciler/source/resources/receive_adapter.go
@@ -24,6 +24,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	v1alpha1 "knative.dev/eventing-rabbitmq/pkg/apis/sources/v1alpha1"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
 )
 
@@ -150,6 +151,32 @@ func MakeReceiveAdapter(args *ReceiveAdapterArgs) *v1.Deployment {
 		},
 	}
 
+	if args.Source.Spec.Retry != nil {
+		env = append(env, corev1.EnvVar{
+			Name:  "HTTP_SENDER_RETRY",
+			Value: strconv.FormatInt(int64(*args.Source.Spec.Retry), 10),
+		})
+	}
+
+	if args.Source.Spec.BackoffPolicy != nil {
+		env = append(env, corev1.EnvVar{
+			Name:  "HTTP_SENDER_BACKOFF_POLICY",
+			Value: string(*args.Source.Spec.BackoffPolicy),
+		})
+	} else {
+		env = append(env, corev1.EnvVar{
+			Name:  "HTTP_SENDER_BACKOFF_POLICY",
+			Value: string(eventingduckv1.BackoffPolicyExponential),
+		})
+	}
+
+	if args.Source.Spec.BackoffDelay != nil {
+		env = append(env, corev1.EnvVar{
+			Name:  "HTTP_SENDER_BACKOFF_DELAY",
+			Value: *args.Source.Spec.BackoffDelay,
+		})
+	}
+
 	return &v1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:         kmeta.ChildName(fmt.Sprintf("rabbitmqsource-%s-", args.Source.Name), string(args.Source.UID)),
diff --git a/pkg/reconciler/trigger/resources/dispatcher.go b/pkg/reconciler/trigger/resources/dispatcher.go
index 58bf3a7c..f64bd5d6 100644
--- a/pkg/reconciler/trigger/resources/dispatcher.go
+++ b/pkg/reconciler/trigger/resources/dispatcher.go
@@ -35,7 +35,7 @@ import (
 
 const (
 	dispatcherContainerName = "dispatcher"
-	prefetchAnnotation      = "rabbitmq.eventing.knative.dev/prefetchCount"
+	PrefetchAnnotation      = "rabbitmq.eventing.knative.dev/prefetchCount"
 )
 
 // DispatcherArgs are the arguments to create a dispatcher deployment. There's
@@ -117,7 +117,7 @@ func MakeDispatcherDeployment(args *DispatcherArgs) *appsv1.Deployment {
 				})
 		}
 	}
-	if prefetch, ok := args.Trigger.ObjectMeta.Annotations[prefetchAnnotation]; ok {
+	if prefetch, ok := args.Trigger.ObjectMeta.Annotations[PrefetchAnnotation]; ok {
 		dispatcher.Env = append(dispatcher.Env,
 			corev1.EnvVar{
 				Name:  "PREFETCH_COUNT",
diff --git a/pkg/reconciler/trigger/trigger.go b/pkg/reconciler/trigger/trigger.go
index 7f0e1dc0..ec0382f5 100644
--- a/pkg/reconciler/trigger/trigger.go
+++ b/pkg/reconciler/trigger/trigger.go
@@ -307,7 +307,7 @@ func (r *Reconciler) reconcileDeployment(ctx context.Context, d *v1.Deployment)
 		return d, nil
 	} else if err != nil {
 		return nil, err
-	} else if !equality.Semantic.DeepDerivative(d.Spec.Template, current.Spec.Template) {
+	} else if !equality.Semantic.DeepEqual(d.Spec.Template, current.Spec.Template) {
 		// Don't modify the informers copy.
 		desired := current.DeepCopy()
 		desired.Spec = d.Spec
diff --git a/source/source-example.yaml b/source/source-example.yaml
index 2c400754..15f1b024 100644
--- a/source/source-example.yaml
+++ b/source/source-example.yaml
@@ -6,6 +6,9 @@ metadata:
 spec:
   brokers: RABBITMQ_BROKERS
   topic: RABBITMQ_TOPIC
+  retry: 5
+  backoffPolicy: "linear"
+  backoffDelay: "50ms"
   user:
     secretKeyRef:
       name: RABBITMQ_USER


Test patch:
diff --git a/pkg/adapter/adapter_test.go b/pkg/adapter/adapter_test.go
index 308c7307..72655e5f 100644
--- a/pkg/adapter/adapter_test.go
+++ b/pkg/adapter/adapter_test.go
@@ -163,7 +163,6 @@ func TestPostMessage_ServeHTTP(t *testing.T) {
 			}
 
 			err = a.postMessage(m)
-
 			if tc.error && err == nil {
 				t.Errorf("expected error, but got %v", err)
 			}
diff --git a/pkg/reconciler/brokerstandalone/broker_test.go b/pkg/reconciler/brokerstandalone/broker_test.go
index 2800c8f1..7c5c17a9 100644
--- a/pkg/reconciler/brokerstandalone/broker_test.go
+++ b/pkg/reconciler/brokerstandalone/broker_test.go
@@ -86,9 +86,7 @@ const (
 
 	dispatcherImage = "dispatcherimage"
 
-	bindingList = `
-[
-  {
+	bindingList = `[{
     "source": "knative-test-broker",
     "vhost": "/",
     "destination": "test-namespace-test-broker",
@@ -100,9 +98,7 @@ const (
       "type": "dev.knative.sources.ping"
     },
     "properties_key": "test-namespace-test-broker"
-  }
-]
-`
+  }]`
 )
 
 var (
@@ -868,7 +864,7 @@ func TestReconcile(t *testing.T) {
 		fakeServer.Start()
 		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			w.WriteHeader(http.StatusCreated)
-			fmt.Fprintln(w, bindingList)
+			fmt.Fprint(w, bindingList)
 		}))
 		r := &Reconciler{
 			eventingClientSet:  fakeeventingclient.Get(ctx),
diff --git a/pkg/reconciler/source/resources/receive_adapter_test.go b/pkg/reconciler/source/resources/receive_adapter_test.go
index 15ab27da..0ce4aa49 100644
--- a/pkg/reconciler/source/resources/receive_adapter_test.go
+++ b/pkg/reconciler/source/resources/receive_adapter_test.go
@@ -24,238 +24,279 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	v1alpha12 "knative.dev/eventing-rabbitmq/pkg/apis/sources/v1alpha1"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 )
 
 func TestMakeReceiveAdapter(t *testing.T) {
+	var retry int32 = 5
 	prefetchCount := 10
-	src := &v1alpha12.RabbitmqSource{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      "source-name",
-			Namespace: "source-namespace",
-		},
-		Spec: v1alpha12.RabbitmqSourceSpec{
-			ServiceAccountName: "source-svc-acct",
-			Topic:              "topic",
-			Brokers:            "amqp://guest:guest@localhost:5672/",
-			User: v1alpha12.SecretValueFromSource{
-				SecretKeyRef: &corev1.SecretKeySelector{
-					LocalObjectReference: corev1.LocalObjectReference{
-						Name: "the-user-secret",
-					},
-					Key: "user",
+	backoffDelay := "50ms"
+
+	for _, tt := range []struct {
+		name          string
+		backoffPolicy eventingduckv1.BackoffPolicyType
+	}{{
+		name:          "Backoff policy linear",
+		backoffPolicy: eventingduckv1.BackoffPolicyLinear,
+	},
+		{
+			name:          "Backoff policy exponential",
+			backoffPolicy: eventingduckv1.BackoffPolicyExponential,
+		}} {
+		t.Run(tt.name, func(t *testing.T) {
+			tt := tt
+			t.Parallel()
+
+			src := &v1alpha12.RabbitmqSource{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "source-name",
+					Namespace: "source-namespace",
 				},
-			},
-			Password: v1alpha12.SecretValueFromSource{
-				SecretKeyRef: &corev1.SecretKeySelector{
-					LocalObjectReference: corev1.LocalObjectReference{
-						Name: "the-password-secret",
+				Spec: v1alpha12.RabbitmqSourceSpec{
+					ServiceAccountName: "source-svc-acct",
+					Topic:              "topic",
+					Brokers:            "amqp://guest:guest@localhost:5672/",
+					User: v1alpha12.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{
+								Name: "the-user-secret",
+							},
+							Key: "user",
+						},
 					},
-					Key: "password",
+					Password: v1alpha12.SecretValueFromSource{
+						SecretKeyRef: &corev1.SecretKeySelector{
+							LocalObjectReference: corev1.LocalObjectReference{
+								Name: "the-password-secret",
+							},
+							Key: "password",
+						},
+					},
+					Predeclared: true,
+					ExchangeConfig: v1alpha12.RabbitmqSourceExchangeConfigSpec{
+						Name:       "logs",
+						TypeOf:     "topic",
+						Durable:    true,
+						AutoDelete: false,
+						Internal:   false,
+						NoWait:     false,
+					},
+					QueueConfig: v1alpha12.RabbitmqSourceQueueConfigSpec{
+						Name:       "",
+						RoutingKey: "*.critical",
+						Durable:    false,
+						AutoDelete: false,
+						Exclusive:  false,
+						NoWait:     false,
+					},
+					ChannelConfig: v1alpha12.RabbitmqChannelConfigSpec{
+						PrefetchCount: &prefetchCount,
+					},
+					Retry:         &retry,
+					BackoffDelay:  &backoffDelay,
+					BackoffPolicy: &tt.backoffPolicy,
 				},
-			},
-			Predeclared: true,
-			ExchangeConfig: v1alpha12.RabbitmqSourceExchangeConfigSpec{
-				Name:       "logs",
-				TypeOf:     "topic",
-				Durable:    true,
-				AutoDelete: false,
-				Internal:   false,
-				NoWait:     false,
-			},
-			QueueConfig: v1alpha12.RabbitmqSourceQueueConfigSpec{
-				Name:       "",
-				RoutingKey: "*.critical",
-				Durable:    false,
-				AutoDelete: false,
-				Exclusive:  false,
-				NoWait:     false,
-			},
-			ChannelConfig: v1alpha12.RabbitmqChannelConfigSpec{
-				PrefetchCount: &prefetchCount,
-			},
-		},
-	}
+			}
 
-	got := MakeReceiveAdapter(&ReceiveAdapterArgs{
-		Image:  "test-image",
-		Source: src,
-		Labels: map[string]string{
-			"test-key1": "test-value1",
-			"test-key2": "test-value2",
-		},
-		SinkURI: "sink-uri",
-	})
-
-	one := int32(1)
-	want := &v1.Deployment{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:         "rabbitmqsource-source-name-",
-			Namespace:    "source-namespace",
-			GenerateName: "source-name-",
-			Labels: map[string]string{
-				"test-key1": "test-value1",
-				"test-key2": "test-value2",
-			},
-			OwnerReferences: []metav1.OwnerReference{
-				{
-					APIVersion:         "sources.knative.dev/v1alpha1",
-					Kind:               "RabbitmqSource",
-					Name:               "source-name",
-					Controller:         &[]bool{true}[0],
-					BlockOwnerDeletion: &[]bool{true}[0],
-				},
-			},
-		},
-		Spec: v1.DeploymentSpec{
-			Selector: &metav1.LabelSelector{
-				MatchLabels: map[string]string{
+			got := MakeReceiveAdapter(&ReceiveAdapterArgs{
+				Image:  "test-image",
+				Source: src,
+				Labels: map[string]string{
 					"test-key1": "test-value1",
 					"test-key2": "test-value2",
 				},
-			},
-			Replicas: &one,
-			Template: corev1.PodTemplateSpec{
+				SinkURI: "sink-uri",
+			})
+
+			boPolicy := tt.backoffPolicy
+			one := int32(1)
+			if boPolicy == "" {
+				boPolicy = eventingduckv1.BackoffPolicyExponential
+			}
+
+			want := &v1.Deployment{
 				ObjectMeta: metav1.ObjectMeta{
-					Annotations: map[string]string{
-						"sidecar.istio.io/inject": "true",
-					},
+					Name:         "rabbitmqsource-source-name-",
+					Namespace:    "source-namespace",
+					GenerateName: "source-name-",
 					Labels: map[string]string{
 						"test-key1": "test-value1",
 						"test-key2": "test-value2",
 					},
-				},
-				Spec: corev1.PodSpec{
-					ServiceAccountName: "source-svc-acct",
-					Containers: []corev1.Container{
+					OwnerReferences: []metav1.OwnerReference{
 						{
-							Name:            "receive-adapter",
-							Image:           "test-image",
-							ImagePullPolicy: "IfNotPresent",
-							Env: []corev1.EnvVar{
-								{
-									Name:  "RABBITMQ_BROKERS",
-									Value: "amqp://guest:guest@localhost:5672/",
-								},
-								{
-									Name:  "RABBITMQ_TOPIC",
-									Value: "topic",
-								},
+							APIVersion:         "sources.knative.dev/v1alpha1",
+							Kind:               "RabbitmqSource",
+							Name:               "source-name",
+							Controller:         &[]bool{true}[0],
+							BlockOwnerDeletion: &[]bool{true}[0],
+						},
+					},
+				},
+				Spec: v1.DeploymentSpec{
+					Selector: &metav1.LabelSelector{
+						MatchLabels: map[string]string{
+							"test-key1": "test-value1",
+							"test-key2": "test-value2",
+						},
+					},
+					Replicas: &one,
+					Template: corev1.PodTemplateSpec{
+						ObjectMeta: metav1.ObjectMeta{
+							Annotations: map[string]string{
+								"sidecar.istio.io/inject": "true",
+							},
+							Labels: map[string]string{
+								"test-key1": "test-value1",
+								"test-key2": "test-value2",
+							},
+						},
+						Spec: corev1.PodSpec{
+							ServiceAccountName: "source-svc-acct",
+							Containers: []corev1.Container{
 								{
-									Name: "RABBITMQ_USER",
-									ValueFrom: &corev1.EnvVarSource{
-										SecretKeyRef: &corev1.SecretKeySelector{
-											LocalObjectReference: corev1.LocalObjectReference{
-												Name: "the-user-secret",
+									Name:            "receive-adapter",
+									Image:           "test-image",
+									ImagePullPolicy: "IfNotPresent",
+									Env: []corev1.EnvVar{
+										{
+											Name:  "RABBITMQ_BROKERS",
+											Value: "amqp://guest:guest@localhost:5672/",
+										},
+										{
+											Name:  "RABBITMQ_TOPIC",
+											Value: "topic",
+										},
+										{
+											Name: "RABBITMQ_USER",
+											ValueFrom: &corev1.EnvVarSource{
+												SecretKeyRef: &corev1.SecretKeySelector{
+													LocalObjectReference: corev1.LocalObjectReference{
+														Name: "the-user-secret",
+													},
+													Key: "user",
+												},
 											},
-											Key: "user",
 										},
-									},
-								},
-								{
-									Name: "RABBITMQ_PASSWORD",
-									ValueFrom: &corev1.EnvVarSource{
-										SecretKeyRef: &corev1.SecretKeySelector{
-											LocalObjectReference: corev1.LocalObjectReference{
-												Name: "the-password-secret",
+										{
+											Name: "RABBITMQ_PASSWORD",
+											ValueFrom: &corev1.EnvVarSource{
+												SecretKeyRef: &corev1.SecretKeySelector{
+													LocalObjectReference: corev1.LocalObjectReference{
+														Name: "the-password-secret",
+													},
+													Key: "password",
+												},
 											},
-											Key: "password",
+										},
+										{
+											Name:  "RABBITMQ_ROUTING_KEY",
+											Value: "*.critical",
+										},
+										{
+											Name:  "RABBITMQ_CHANNEL_CONFIG_QOS_GLOBAL",
+											Value: "false",
+										},
+										{
+											Name:  "RABBITMQ_CHANNEL_CONFIG_PREFETCH_COUNT",
+											Value: "10",
+										},
+										{
+											Name:  "RABBITMQ_EXCHANGE_CONFIG_NAME",
+											Value: "logs",
+										},
+										{
+											Name:  "RABBITMQ_EXCHANGE_CONFIG_TYPE",
+											Value: "topic",
+										},
+										{
+											Name:  "RABBITMQ_EXCHANGE_CONFIG_DURABLE",
+											Value: "true",
+										},
+										{
+											Name:  "RABBITMQ_EXCHANGE_CONFIG_AUTO_DELETE",
+											Value: "false",
+										},
+										{
+											Name:  "RABBITMQ_EXCHANGE_CONFIG_INTERNAL",
+											Value: "false",
+										},
+										{
+											Name:  "RABBITMQ_EXCHANGE_CONFIG_NOWAIT",
+											Value: "false",
+										},
+										{
+											Name:  "RABBITMQ_QUEUE_CONFIG_NAME",
+											Value: "",
+										},
+										{
+											Name:  "RABBITMQ_QUEUE_CONFIG_DURABLE",
+											Value: "false",
+										},
+										{
+											Name:  "RABBITMQ_QUEUE_CONFIG_AUTO_DELETE",
+											Value: "false",
+										},
+										{
+											Name:  "RABBITMQ_QUEUE_CONFIG_EXCLUSIVE",
+											Value: "false",
+										},
+										{
+											Name:  "RABBITMQ_QUEUE_CONFIG_NOWAIT",
+											Value: "false",
+										},
+										{
+											Name:  "RABBITMQ_PREDECLARED",
+											Value: "true",
+										},
+										{
+											Name:  "SINK_URI",
+											Value: "sink-uri",
+										},
+										{
+											Name:  "K_SINK",
+											Value: "sink-uri",
+										},
+										{
+											Name:  "NAME",
+											Value: "source-name",
+										},
+										{
+											Name:  "NAMESPACE",
+											Value: "source-namespace",
+										},
+										{
+											Name: "K_LOGGING_CONFIG",
+										},
+										{
+											Name: "K_METRICS_CONFIG",
+										},
+										{
+											Name: "RABBITMQ_VHOST",
+										},
+										{
+											Name:  "HTTP_SENDER_RETRY",
+											Value: "5",
+										},
+										{
+											Name:  "HTTP_SENDER_BACKOFF_POLICY",
+											Value: string(boPolicy),
+										},
+										{
+											Name:  "HTTP_SENDER_BACKOFF_DELAY",
+											Value: "50ms",
 										},
 									},
 								},
-								{
-									Name:  "RABBITMQ_ROUTING_KEY",
-									Value: "*.critical",
-								},
-								{
-									Name:  "RABBITMQ_CHANNEL_CONFIG_QOS_GLOBAL",
-									Value: "false",
-								},
-								{
-									Name:  "RABBITMQ_CHANNEL_CONFIG_PREFETCH_COUNT",
-									Value: "10",
-								},
-								{
-									Name:  "RABBITMQ_EXCHANGE_CONFIG_NAME",
-									Value: "logs",
-								},
-								{
-									Name:  "RABBITMQ_EXCHANGE_CONFIG_TYPE",
-									Value: "topic",
-								},
-								{
-									Name:  "RABBITMQ_EXCHANGE_CONFIG_DURABLE",
-									Value: "true",
-								},
-								{
-									Name:  "RABBITMQ_EXCHANGE_CONFIG_AUTO_DELETE",
-									Value: "false",
-								},
-								{
-									Name:  "RABBITMQ_EXCHANGE_CONFIG_INTERNAL",
-									Value: "false",
-								},
-								{
-									Name:  "RABBITMQ_EXCHANGE_CONFIG_NOWAIT",
-									Value: "false",
-								},
-								{
-									Name:  "RABBITMQ_QUEUE_CONFIG_NAME",
-									Value: "",
-								},
-								{
-									Name:  "RABBITMQ_QUEUE_CONFIG_DURABLE",
-									Value: "false",
-								},
-								{
-									Name:  "RABBITMQ_QUEUE_CONFIG_AUTO_DELETE",
-									Value: "false",
-								},
-								{
-									Name:  "RABBITMQ_QUEUE_CONFIG_EXCLUSIVE",
-									Value: "false",
-								},
-								{
-									Name:  "RABBITMQ_QUEUE_CONFIG_NOWAIT",
-									Value: "false",
-								},
-								{
-									Name:  "RABBITMQ_PREDECLARED",
-									Value: "true",
-								},
-								{
-									Name:  "SINK_URI",
-									Value: "sink-uri",
-								},
-								{
-									Name:  "K_SINK",
-									Value: "sink-uri",
-								},
-								{
-									Name:  "NAME",
-									Value: "source-name",
-								},
-								{
-									Name:  "NAMESPACE",
-									Value: "source-namespace",
-								},
-								{
-									Name: "K_LOGGING_CONFIG",
-								},
-								{
-									Name: "K_METRICS_CONFIG",
-								},
-								{
-									Name: "RABBITMQ_VHOST",
-								},
 							},
 						},
 					},
 				},
-			},
-		},
-	}
+			}
 
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("unexpected deploy (-want, +got) = %v", diff)
+			if diff := cmp.Diff(want, got); diff != "" {
+				t.Errorf("unexpected deploy (-want, +got) = %v", diff)
+			}
+		})
 	}
 }
diff --git a/pkg/reconciler/trigger/resources/dispatcher_test.go b/pkg/reconciler/trigger/resources/dispatcher_test.go
index 14e743de..f37ba1ce 100644
--- a/pkg/reconciler/trigger/resources/dispatcher_test.go
+++ b/pkg/reconciler/trigger/resources/dispatcher_test.go
@@ -208,9 +208,9 @@ func withDLX(args *DispatcherArgs) {
 func withPrefetch(c string) func(*DispatcherArgs) {
 	return func(args *DispatcherArgs) {
 		if args.Trigger.ObjectMeta.Annotations == nil {
-			args.Trigger.ObjectMeta.Annotations = map[string]string{prefetchAnnotation: c}
+			args.Trigger.ObjectMeta.Annotations = map[string]string{PrefetchAnnotation: c}
 		} else {
-			args.Trigger.ObjectMeta.Annotations[prefetchAnnotation] = c
+			args.Trigger.ObjectMeta.Annotations[PrefetchAnnotation] = c
 		}
 	}
 }
diff --git a/pkg/reconciler/trigger/trigger_test.go b/pkg/reconciler/trigger/trigger_test.go
index f4220395..c9bd66ad 100644
--- a/pkg/reconciler/trigger/trigger_test.go
+++ b/pkg/reconciler/trigger/trigger_test.go
@@ -703,6 +703,35 @@ func TestReconcile(t *testing.T) {
 					WithTriggerDependencyReady(),
 				),
 			}},
+		}, {
+			Name: "Deployment updated when prefetch count removed",
+			Key:  testKey,
+			Objects: []runtime.Object{
+				ReadyBroker(),
+				NewTrigger(triggerName, testNS, brokerName,
+					WithTriggerUID(triggerUID),
+					WithTriggerSubscriberURI(subscriberURI)),
+				createSecret(rabbitURL),
+				markReady(createQueue()),
+				markReady(createPolicy(true)),
+				markReady(createBinding(false)),
+				createDispatcherDeploymentWithPrefetchCount(),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{{
+				Object: createDispatcherDeployment(),
+			}},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+				Object: NewTrigger(triggerName, testNS, brokerName,
+					WithTriggerUID(triggerUID),
+					WithTriggerSubscriberURI(subscriberURI),
+					WithInitTriggerConditions,
+					WithTriggerBrokerReady(),
+					WithTriggerDeadLetterSinkNotConfigured(),
+					WithTriggerSubscribed(),
+					WithTriggerDependencyReady(),
+					WithTriggerSubscriberResolvedSucceeded(),
+					WithTriggerStatusSubscriberURI(subscriberURI)),
+			}},
 		},
 	}
 
@@ -985,6 +1014,31 @@ func createDifferentDispatcherDeployment() *appsv1.Deployment {
 	return resources.MakeDispatcherDeployment(args)
 }
 
+func createDispatcherDeploymentWithPrefetchCount() *appsv1.Deployment {
+	args := &resources.DispatcherArgs{
+		Trigger: &eventingv1.Trigger{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      triggerName,
+				Namespace: testNS,
+				UID:       triggerUID,
+				Annotations: map[string]string{
+					resources.PrefetchAnnotation: "10",
+				},
+			},
+			Spec: eventingv1.TriggerSpec{
+				Broker: brokerName,
+			},
+		},
+		Image:              dispatcherImage,
+		RabbitMQSecretName: rabbitSecretName,
+		QueueName:          queueName,
+		BrokerUrlSecretKey: "brokerURL",
+		BrokerIngressURL:   brokerAddress,
+		Subscriber:         subscriberAddress,
+	}
+	return resources.MakeDispatcherDeployment(args)
+}
+
 func makeSubscriberAddressableAsUnstructured() *unstructured.Unstructured {
 	return &unstructured.Unstructured{
 		Object: map[string]interface{}{
diff --git a/pkg/reconciler/triggerstandalone/trigger_test.go b/pkg/reconciler/triggerstandalone/trigger_test.go
index c8d7d030..6b279a05 100644
--- a/pkg/reconciler/triggerstandalone/trigger_test.go
+++ b/pkg/reconciler/triggerstandalone/trigger_test.go
@@ -101,9 +101,7 @@ const (
 	subscriberGroup   = "serving.knative.dev"
 	subscriberVersion = "v1"
 
-	bindingList = `
-[
-  {
+	bindingList = `[{
     "source": "knative-test-broker",
     "vhost": "/",
     "destination": "test-namespace-test-broker",
@@ -115,9 +113,7 @@ const (
       "type": "dev.knative.sources.ping"
     },
     "properties_key": "test-namespace-test-broker"
-  }
-]
-`
+  }]`
 )
 
 var (
@@ -712,7 +708,7 @@ func TestReconcile(t *testing.T) {
 		fakeServer.Start()
 		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			w.WriteHeader(http.StatusCreated)
-			fmt.Fprintln(w, bindingList)
+			fmt.Fprint(w, bindingList)
 		}))
 		r := &Reconciler{
 			eventingClientSet:  fakeeventingclient.Get(ctx),
diff --git a/test/README.md b/test/README.md
index 90c1a643..924fa9bc 100644
--- a/test/README.md
+++ b/test/README.md
@@ -14,9 +14,9 @@ components and running tests - everything done with `make`.
 make .envrc && . .envrc
 ```
 
-Here we create our environment file containing patched PATH, various exports, etc.
-I personally like to use [direnv](https://direnv.net/) too. It will pick up .envrc and source it
-each time I cd'ing to my Eventng RabbitMQ local copy and more...
+Here we create our environment file containing patched `PATH`, various exports, etc.
+[Direnv](https://direnv.net/) will pick up `.envrc` and source it
+each time shell visits Eventing RabbitMQ folder.
 
 ### Create kind cluster
 
@@ -24,7 +24,7 @@ each time I cd'ing to my Eventng RabbitMQ local copy and more...
 make kind-cluster
 ```
 
-This will install `kind` to ./bin and initialize the cluster. Kind uses [kind.yaml](e2e/kind.yaml) as cluster configuration.
+This will install `kind` to ./bin and initialize the cluster. `Kind` uses [kind.yaml](e2e/kind.yaml) as cluster configuration.
 
 
 ### Install Knative, Certificate manager, RabbitMQ
@@ -39,8 +39,8 @@ make install
 make k9s
 ```
 
-[k9s](https://k9scli.io/) is a "Kubernetes CLI To Manage Your Clusters In Style". That `make` target takes care of k9s installation and configuration.
-This target will run k9s pointed to our kind cluster.
+[k9s](https://k9scli.io/) is a "Kubernetes CLI To Manage Your Clusters In Style". That `make` target takes care of `k9s` installation and configuration.
+This target will run `k9s` pointed to our kind cluster.
 
 ### Running tests
 
@@ -60,8 +60,8 @@ test-unit-uncached                  Run unit tests with no cache
 
 ## Temporary files
 
-Makefile will download and store binaries such as `kind`, `k9s`, `kubectl` to (curdir)/bin.
-Configurations will be stored in (curdir)/.config.
+Makefile will download and store binaries such as `kind`, `k9s`, `kubectl` to `(curdir)/bin`.
+Configurations will be stored in `(curdir)/.config`.
 Both paths are in the `.gitignore`.
 
 ## How to start fresh