Release 1.5 (#839)

* fixed source delivery and adapter sendwithretries retryConfig + fixed source perf test related tests and added retry to one sample

* added retry test to source receive adapter

Author: gabo1208

pkg/adapter/adapter.go

@@ -89,6 +89,10 @@ type Adapter struct {
 
 var _ adapter.MessageAdapter = (*Adapter)(nil)
 var _ adapter.MessageAdapterConstructor = NewAdapter
+var (
+	retryConfig  kncloudevents.RetryConfig = kncloudevents.NoRetries()
+	retriesInt32 int32                     = 0
+)
 
 func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, httpMessageSender *kncloudevents.HTTPMessageSender, reporter source.StatsReporter) adapter.MessageAdapter {
 	logger := logging.FromContext(ctx).Desugar()
@@ -220,9 +224,20 @@ func (a *Adapter) ConsumeMessages(channel *wabbit.Channel,
 func (a *Adapter) PollForMessages(channel *wabbit.Channel,
 	queue *wabbit.Queue, stopCh <-chan struct{}) error {
 	logger := a.logger
+	var err error
+	retriesInt32 = int32(a.config.Retry)
+	backoffPolicy := (v1.BackoffPolicyType)(a.config.BackoffPolicy)
+	backoffDelay := a.config.BackoffDelay.String()
+	retryConfig, err = kncloudevents.RetryConfigFromDeliverySpec(v1.DeliverySpec{
+		BackoffPolicy: &backoffPolicy,
+		BackoffDelay:  &backoffDelay,
+		Retry:         &retriesInt32,
+	})
+	if err != nil {
+		a.logger.Error("error retrieving retryConfig from deliverySpec", zap.Error(err))
+	}
 
 	msgs, _ := a.ConsumeMessages(channel, queue, logger)
-
 	wg := &sync.WaitGroup{}
 	workerCount := a.config.ChannelConfig.Parallelism
 	wg.Add(workerCount)
@@ -291,17 +306,7 @@ func (a *Adapter) postMessage(msg wabbit.Delivery) error {
 		return err
 	}
 
-	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,
-	})
-
+	res, err := a.httpMessageSender.SendWithRetries(req, &retryConfig)
 	if err != nil {
 		a.logger.Error("error while sending the message", zap.Error(err))
 		return err

pkg/adapter/adapter_test.go

@@ -25,6 +25,7 @@ import (
 	"net/http/httptest"
 	"reflect"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -36,34 +37,38 @@ 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"
 )
 
+type handlerFunc func(http.ResponseWriter, *http.Request)
+
 func TestPostMessage_ServeHTTP(t *testing.T) {
 	testCases := map[string]struct {
-		sink                       func(http.ResponseWriter, *http.Request)
+		handlers                   []handlerFunc
 		reqBody, expectedEventType string
 		reqHeaders                 http.Header
 		data                       map[string]interface{}
 		headers                    wabbit.Option
 		attributes                 map[string]string
 		withMsgId, isCe, error     bool
+		retry                      int
 	}{
 		"accepted": {
-			sink:    sinkAccepted,
-			reqBody: `{"key":"value"}`,
-			data:    map[string]interface{}{"key": "value"},
+			handlers: []handlerFunc{sinkAccepted},
+			reqBody:  `{"key":"value"}`,
+			data:     map[string]interface{}{"key": "value"},
 		},
 		"accepted with msg id": {
-			sink:      sinkAccepted,
+			handlers:  []handlerFunc{sinkAccepted},
 			reqBody:   `{"key":"value"}`,
 			withMsgId: true,
 			data:      map[string]interface{}{"key": "value"},
 		},
 		"accepted with binary cloudevent": {
-			sink:      sinkAccepted,
+			handlers:  []handlerFunc{sinkAccepted},
 			reqBody:   `{"test":"test"}`,
 			withMsgId: true,
 			reqHeaders: http.Header{
@@ -82,7 +87,7 @@ func TestPostMessage_ServeHTTP(t *testing.T) {
 			isCe: true,
 		},
 		"accepted with structured cloudevent": {
-			sink: sinkAccepted,
+			handlers: []handlerFunc{sinkAccepted},
 			reqBody: `{"specversion":"1.0","id":1234,` +
 				`"type":"dev.knative.rabbitmq.event","source":"example/source.uri",` +
 				`"content-type":"text/plain","data":"test"}`,
@@ -99,17 +104,24 @@ func TestPostMessage_ServeHTTP(t *testing.T) {
 			isCe: true,
 		},
 		"rejected": {
-			sink:    sinkRejected,
-			reqBody: `{"key":"value"}`,
-			error:   true,
-			data:    map[string]interface{}{"key": "value"},
+			handlers: []handlerFunc{sinkRejected},
+			reqBody:  `{"key":"value"}`,
+			error:    true,
+			data:     map[string]interface{}{"key": "value"},
+		},
+		"retried 3 times succesfull on the 4th ": {
+			retry:    5,
+			handlers: []handlerFunc{sinkRejected, sinkRejected, sinkRejected, sinkAccepted},
+			reqBody:  `{"key":"value"}`,
+			error:    false,
+			data:     map[string]interface{}{"key": "value"},
 		},
 	}
 
 	for n, tc := range testCases {
 		t.Run(n, func(t *testing.T) {
 			h := &fakeHandler{
-				handler: tc.sink,
+				handlers: tc.handlers,
 			}
 			sinkServer := httptest.NewServer(h)
 			defer sinkServer.Close()
@@ -120,21 +132,12 @@ func TestPostMessage_ServeHTTP(t *testing.T) {
 			}
 
 			statsReporter, _ := source.NewStatsReporter()
-
+			config := adapterConfig{}
+			if tc.retry > 0 {
+				config = adapterConfig{Retry: tc.retry, BackoffPolicy: string(v1.BackoffPolicyLinear), BackoffDelay: time.Duration(10000)}
+			}
 			a := &Adapter{
-				config: &adapterConfig{
-					Broker: "amqp://guest:guest@localhost:5672/",
-					ExchangeConfig: ExchangeConfig{
-						Type:       "topic",
-						Durable:    true,
-						AutoDelete: false,
-					},
-					QueueConfig: QueueConfig{
-						Name:       "",
-						Durable:    false,
-						AutoDelete: false,
-					},
-				},
+				config:            &config,
 				context:           context.TODO(),
 				httpMessageSender: s,
 				logger:            zap.NewNop(),
@@ -337,11 +340,15 @@ func TestAdapter_StartAmqpClient(t *testing.T) {
 type fakeHandler struct {
 	body   []byte
 	header http.Header
+	mu     sync.Mutex
 
-	handler func(http.ResponseWriter, *http.Request)
+	receiveCount int
+	handlers     []handlerFunc
 }
 
 func (h *fakeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
 	h.header = r.Header
 	body, err := ioutil.ReadAll(r.Body)
 	if err != nil {
@@ -351,7 +358,8 @@ func (h *fakeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	h.body = body
 
 	defer r.Body.Close()
-	h.handler(w, r)
+	h.receiveCount++
+	h.handlers[h.receiveCount](w, r)
 }
 
 func sinkAccepted(writer http.ResponseWriter, req *http.Request) {
@@ -502,7 +510,7 @@ func TestAdapter_NewAdapter(t *testing.T) {
 	ctx := context.TODO()
 	env := NewEnvConfig()
 	h := &fakeHandler{
-		handler: sinkAccepted,
+		handlers: []handlerFunc{sinkAccepted},
 	}
 
 	sinkServer := httptest.NewServer(h)

samples/source/external-cluster/200-perf-test.yaml

@@ -55,13 +55,12 @@ spec:
               value: "true"
             - name: SIZE
               value: "1000"
-
             - name: EXCHANGE
               value: "eventing-rabbitmq-source"
             - name: TYPE
-              value: "fanout"
+              value: "headers"
             # This version of PerfTest cannot declare a durable exchange: https://github.com/rabbitmq/rabbitmq-perf-test/issues/281
-            # This version of PerfTest will create an exchange if it does not exist if if PREDECLARED is set to true: https://github.com/rabbitmq/rabbitmq-perf-test/issues/282
+            # This version of PerfTest will create an exchange if it does not exist if PREDECLARED is set to true: https://github.com/rabbitmq/rabbitmq-perf-test/issues/282
             # This means that we cannot use the messaging topology operator to manage our exchange as it can create race conditions with PerfTest: https://github.com/rabbitmq/messaging-topology-operator
             # We have chosen the "lesser evil" and are using a non-durable exchange
 

samples/source/quick-setup/300-perf-test.yaml

@@ -55,11 +55,12 @@ spec:
               value: "true"
             - name: SIZE
               value: "1000"
-
+            - name: PREDECLARED
+              value: "true"
             - name: EXCHANGE
               value: "eventing-rabbitmq-source"
             - name: TYPE
-              value: "fanout"
+              value: "headers"
             # This version of PerfTest cannot declare a durable exchange: https://github.com/rabbitmq/rabbitmq-perf-test/issues/281
             # This version of PerfTest will create an exchange if it does not exist if if PREDECLARED is set to true: https://github.com/rabbitmq/rabbitmq-perf-test/issues/282
             # This means that we cannot use the messaging topology operator to manage our exchange as it can create race conditions with PerfTest: https://github.com/rabbitmq/messaging-topology-operator