Fix issue knative#4425 properly creating the request using the right …

…constructor (knative#4426)

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

pkg/kncloudevents/message_sender.go

@@ -142,7 +142,12 @@ func (s *HTTPMessageSender) SendWithRetries(req *nethttp.Request, config *RetryC
 		},
 	}
 
-	return retryableClient.Do(&retryablehttp.Request{Request: req})
+	retryableReq, err := retryablehttp.FromRequest(req)
+	if err != nil {
+		return nil, err
+	}
+
+	return retryableClient.Do(retryableReq)
 }
 
 func NoRetries() RetryConfig {

pkg/kncloudevents/message_sender_test.go

@@ -20,16 +20,21 @@ import (
 	"context"
 	"net/http"
 	"net/http/httptest"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/cloudevents/sdk-go/v2/binding/buffering"
+	bindingtest "github.com/cloudevents/sdk-go/v2/binding/test"
+	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
+	cetest "github.com/cloudevents/sdk-go/v2/test"
 	"github.com/stretchr/testify/assert"
-	"go.uber.org/atomic"
 	"k8s.io/utils/pointer"
 
+	"knative.dev/pkg/ptr"
+
 	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
-	"knative.dev/pkg/ptr"
 )
 
 // Test The RetryConfigFromDeliverySpec() Functionality
@@ -149,9 +154,9 @@ func TestHTTPMessageSenderSendWithRetries(t *testing.T) {
 	}}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			var n atomic.Int32
+			var n int32
 			server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
-				n.Inc()
+				atomic.AddInt32(&n, 1)
 				writer.WriteHeader(tt.wantStatus)
 			}))
 
@@ -168,13 +173,64 @@ func TestHTTPMessageSenderSendWithRetries(t *testing.T) {
 			if got.StatusCode != http.StatusServiceUnavailable {
 				t.Fatalf("SendWithRetries() got = %v, want %v", got.StatusCode, http.StatusServiceUnavailable)
 			}
-			if count := int(n.Load()); count != tt.wantDispatch {
+			if count := int(atomic.LoadInt32(&n)); count != tt.wantDispatch {
 				t.Fatalf("expected %d retries got %d", tt.config.RetryMax, count)
 			}
 		})
 	}
 }
 
+func TestHTTPMessageSenderSendWithRetriesWithBufferedMessage(t *testing.T) {
+	t.Parallel()
+
+	const wantToSkip = 9
+	config := &RetryConfig{
+		RetryMax: wantToSkip,
+		CheckRetry: func(ctx context.Context, resp *http.Response, err error) (bool, error) {
+			return true, nil
+		},
+		Backoff: func(attemptNum int, resp *http.Response) time.Duration {
+			return time.Millisecond * 50 * time.Duration(attemptNum)
+		},
+	}
+
+	var n uint32
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		thisReqN := atomic.AddUint32(&n, 1)
+		if thisReqN <= wantToSkip {
+			writer.WriteHeader(http.StatusServiceUnavailable)
+		} else {
+			writer.WriteHeader(http.StatusAccepted)
+		}
+	}))
+
+	sender := &HTTPMessageSender{
+		Client: http.DefaultClient,
+	}
+
+	request, err := http.NewRequest("POST", server.URL, nil)
+	assert.Nil(t, err)
+
+	// Create a message similar to the one we send with channels
+	mockMessage := bindingtest.MustCreateMockBinaryMessage(cetest.FullEvent())
+	bufferedMessage, err := buffering.BufferMessage(context.TODO(), mockMessage)
+	assert.Nil(t, err)
+
+	err = cehttp.WriteRequest(context.TODO(), bufferedMessage, request)
+	assert.Nil(t, err)
+
+	got, err := sender.SendWithRetries(request, config)
+	if err != nil {
+		t.Fatalf("SendWithRetries() error = %v, wantErr nil", err)
+	}
+	if got.StatusCode != http.StatusAccepted {
+		t.Fatalf("SendWithRetries() got = %v, want %v", got.StatusCode, http.StatusAccepted)
+	}
+	if count := atomic.LoadUint32(&n); count != wantToSkip+1 {
+		t.Fatalf("expected %d count got %d", wantToSkip+1, count)
+	}
+}
+
 func TestRetryConfigFromDeliverySpecCheckRetry(t *testing.T) {
 	const retryMax = 10
 	linear := eventingduck.BackoffPolicyLinear