Repository: knative-sandbox/eventing-kafka
Pull request #1345 (2023-06-01 08:01:10)

Update dependencies and fix compile issues
Dependency bump as #1344 + fixing the compile issues

creydr: /assign @matzew @pierDipi 
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/1345#" title="Author self-approved">creydr</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/1345#pullrequestreview-1455259006" title="Approved">pierDipi</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).

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/blob/main/OWNERS)~~ [pierDipi]

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/go.mod b/go.mod
index b1c8eb796..f2d1b2566 100644
--- a/go.mod
+++ b/go.mod
@@ -24,10 +24,10 @@ require (
 	k8s.io/apimachinery v0.26.5
 	k8s.io/client-go v0.26.5
 	k8s.io/utils v0.0.0-20221108210102-8e77b1f39fe2
-	knative.dev/eventing v0.37.1-0.20230524182623-6a11c5f02ccf
-	knative.dev/hack v0.0.0-20230524013611-5812c57cc1ac
-	knative.dev/pkg v0.0.0-20230524013411-ec20442c6ead
-	knative.dev/reconciler-test v0.0.0-20230524170754-968cbdf0912e
+	knative.dev/eventing v0.37.1-0.20230531153536-70113e887eec
+	knative.dev/hack v0.0.0-20230530201435-5b7907fbe350
+	knative.dev/pkg v0.0.0-20230531073936-5671699f23d9
+	knative.dev/reconciler-test v0.0.0-20230530092335-cf14f44032b9
 	sigs.k8s.io/yaml v1.3.0
 )
 
diff --git a/go.sum b/go.sum
index 016c05f86..45a3b82ff 100644
--- a/go.sum
+++ b/go.sum
@@ -1085,14 +1085,14 @@ k8s.io/kube-openapi v0.0.0-20221012153701-172d655c2280 h1:+70TFaan3hfJzs+7VK2o+O
 k8s.io/kube-openapi v0.0.0-20221012153701-172d655c2280/go.mod h1:+Axhij7bCpeqhklhUTe3xmOn6bWxolyZEeyaFpjGtl4=
 k8s.io/utils v0.0.0-20221108210102-8e77b1f39fe2 h1:GfD9OzL11kvZN5iArC6oTS7RTj7oJOIfnislxYlqTj8=
 k8s.io/utils v0.0.0-20221108210102-8e77b1f39fe2/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
-knative.dev/eventing v0.37.1-0.20230524182623-6a11c5f02ccf h1:LovZ04fWMgDDwSrHs8A0Ta0rXg7TFdBGeMARV7CuEZo=
-knative.dev/eventing v0.37.1-0.20230524182623-6a11c5f02ccf/go.mod h1:a+FNFGSECC36bVTVgExkHbZYxFKQ2qG2cYELR6HDlwg=
-knative.dev/hack v0.0.0-20230524013611-5812c57cc1ac h1:XQrhtfWPsyuxwSxv4GhcFnqZh75Nq+L1iZWhFMz4aPg=
-knative.dev/hack v0.0.0-20230524013611-5812c57cc1ac/go.mod h1:yk2OjGDsbEnQjfxdm0/HJKS2WqTLEFg/N6nUs6Rqx3Q=
-knative.dev/pkg v0.0.0-20230524013411-ec20442c6ead h1:dcBDhkP6hd9VaV9nLukg75bIrvJs4YiSVOv0h2evOFc=
-knative.dev/pkg v0.0.0-20230524013411-ec20442c6ead/go.mod h1:dqC6IrvyBE7E+oZocs5PkVhq1G59pDTA7r8U17EAKMk=
-knative.dev/reconciler-test v0.0.0-20230524170754-968cbdf0912e h1:Vswp9X9+WPg+56nQA/jr5D0SQLj89zVsknSQNwesL5U=
-knative.dev/reconciler-test v0.0.0-20230524170754-968cbdf0912e/go.mod h1:iB/uRML7hm+V0YvUrAxO+SSq1wg0FRi2p8aVCIjbzeo=
+knative.dev/eventing v0.37.1-0.20230531153536-70113e887eec h1:b3ASwmJm14vffBfkYUpbWnM7/p9bH/DDkHeBJECu6MU=
+knative.dev/eventing v0.37.1-0.20230531153536-70113e887eec/go.mod h1:dV8pq4fMacHc+iWeNGNCr5STK9Y8QiA7B/100zL1Abg=
+knative.dev/hack v0.0.0-20230530201435-5b7907fbe350 h1:lSV9uFTohrQd7i5OUhLnR87PwUHJplQKJSEiZgzP/VY=
+knative.dev/hack v0.0.0-20230530201435-5b7907fbe350/go.mod h1:yk2OjGDsbEnQjfxdm0/HJKS2WqTLEFg/N6nUs6Rqx3Q=
+knative.dev/pkg v0.0.0-20230531073936-5671699f23d9 h1:xd7E2nuQ358cXR4jB7Y5+FH88Mm+SQTZty5pXa1TSfs=
+knative.dev/pkg v0.0.0-20230531073936-5671699f23d9/go.mod h1:dqC6IrvyBE7E+oZocs5PkVhq1G59pDTA7r8U17EAKMk=
+knative.dev/reconciler-test v0.0.0-20230530092335-cf14f44032b9 h1:CzG5Oo639iM2u8x6LrWNznp9b7av7poZV1OpAeJc48s=
+knative.dev/reconciler-test v0.0.0-20230530092335-cf14f44032b9/go.mod h1:iB/uRML7hm+V0YvUrAxO+SSq1wg0FRi2p8aVCIjbzeo=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index cc58d377a..6421c0708 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -35,6 +35,7 @@ import (
 	"go.uber.org/zap"
 
 	"knative.dev/eventing/pkg/metrics/source"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/logging"
 
 	"knative.dev/eventing/pkg/adapter/v2"
@@ -69,12 +70,12 @@ type Adapter struct {
 	config       *AdapterConfig
 	saramaConfig *sarama.Config
 
-	httpMessageSender *kncloudevents.HTTPMessageSender
-	reporter          source.StatsReporter
-	logger            *zap.SugaredLogger
-	keyTypeMapper     func([]byte) interface{}
-	rateLimiter       *rate.Limiter
-	extensions        map[string]string
+	sink          duckv1.Addressable
+	reporter      source.StatsReporter
+	logger        *zap.SugaredLogger
+	keyTypeMapper func([]byte) interface{}
+	rateLimiter   *rate.Limiter
+	extensions    map[string]string
 }
 
 var (
@@ -85,16 +86,16 @@ var (
 	retryConfig                                          = defaultRetryConfig()
 )
 
-func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, httpMessageSender *kncloudevents.HTTPMessageSender, reporter source.StatsReporter) adapter.MessageAdapter {
+func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, sink duckv1.Addressable, reporter source.StatsReporter) adapter.MessageAdapter {
 	logger := logging.FromContext(ctx)
 	config := processed.(*AdapterConfig)
 
 	return &Adapter{
-		config:            config,
-		httpMessageSender: httpMessageSender,
-		reporter:          reporter,
-		logger:            logger,
-		keyTypeMapper:     getKeyTypeMapper(config.KeyType),
+		config:        config,
+		sink:          sink,
+		reporter:      reporter,
+		logger:        logger,
+		keyTypeMapper: getKeyTypeMapper(config.KeyType),
 	}
 }
 func (a *Adapter) GetConsumerGroup() string {
@@ -171,18 +172,18 @@ func (a *Adapter) Handle(ctx context.Context, msg *sarama.ConsumerMessage) (bool
 	ctx, span := tracing.StartTraceFromMessage(a.logger, ctx, message, "kafka-source-"+msg.Topic)
 	defer span.End()
 
-	req, err := a.httpMessageSender.NewCloudEventRequest(ctx)
+	req, err := kncloudevents.NewCloudEventRequest(ctx, a.sink)
 	if err != nil {
 		return false, err
 	}
 
-	err = a.ConsumerMessageToHttpRequest(ctx, msg, req)
+	err = a.ConsumerMessageToHttpRequest(ctx, msg, req.Request)
 	if err != nil {
 		a.logger.Debug("failed to create request", zap.Error(err))
 		return true, err
 	}
 
-	res, err := a.httpMessageSender.SendWithRetries(req, retryConfig)
+	res, err := req.SendWithRetries(retryConfig)
 
 	if err != nil {
 		a.logger.Debug("Error while sending the message", zap.Error(err))
diff --git a/pkg/source/mtadapter/adapter.go b/pkg/source/mtadapter/adapter.go
index d5ce7476f..1b5a185db 100644
--- a/pkg/source/mtadapter/adapter.go
+++ b/pkg/source/mtadapter/adapter.go
@@ -30,8 +30,9 @@ import (
 	"k8s.io/client-go/kubernetes"
 
 	"knative.dev/eventing/pkg/adapter/v2"
-	"knative.dev/eventing/pkg/kncloudevents"
 	"knative.dev/eventing/pkg/metrics/source"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/logging"
 
@@ -188,13 +189,16 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) error {
 		return err
 	}
 
-	httpBindingsSender, err := kncloudevents.NewHTTPMessageSenderWithTarget(obj.Status.SinkURI.String())
+	sinkURI, err := apis.ParseURL(obj.Status.SinkURI.String())
 	if err != nil {
-		a.logger.Errorw("error building cloud event client", zap.Error(err))
+		a.logger.Errorw("error parsing sinkURI", zap.Error(err))
 		return err
 	}
+	sink := duckv1.Addressable{
+		URL: sinkURI,
+	}
 
-	adapter := a.adapterCtor(ctx, &config, httpBindingsSender, reporter)
+	adapter := a.adapterCtor(ctx, &config, sink, reporter)
 
 	// TODO: define Limit interface.
 	if sta, ok := adapter.(*stadapter.Adapter); ok {
diff --git a/vendor/knative.dev/eventing/pkg/adapter/v2/main_message.go b/vendor/knative.dev/eventing/pkg/adapter/v2/main_message.go
index c69c8b3dc..ba8f9ab5c 100644
--- a/vendor/knative.dev/eventing/pkg/adapter/v2/main_message.go
+++ b/vendor/knative.dev/eventing/pkg/adapter/v2/main_message.go
@@ -27,19 +27,19 @@ import (
 	"go.opencensus.io/stats/view"
 	"go.uber.org/zap"
 	"knative.dev/eventing/pkg/metrics/source"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/metrics"
 	"knative.dev/pkg/profiling"
 	"knative.dev/pkg/signals"
-
-	"knative.dev/eventing/pkg/kncloudevents"
 )
 
 type MessageAdapter interface {
 	Start(ctx context.Context) error
 }
 
-type MessageAdapterConstructor func(ctx context.Context, env EnvConfigAccessor, adapter *kncloudevents.HTTPMessageSender, reporter source.StatsReporter) MessageAdapter
+type MessageAdapterConstructor func(ctx context.Context, env EnvConfigAccessor, sink duckv1.Addressable, reporter source.StatsReporter) MessageAdapter
 
 func MainMessageAdapter(component string, ector EnvConfigConstructor, ctor MessageAdapterConstructor) {
 	MainMessageAdapterWithContext(signals.NewContext(), component, ector, ctor)
@@ -108,13 +108,17 @@ func MainMessageAdapterWithContext(ctx context.Context, component string, ector
 	}
 	defer tracer.Shutdown(context.Background())
 
-	httpBindingsSender, err := kncloudevents.NewHTTPMessageSenderWithTarget(env.GetSink())
+	sinkURL, err := apis.ParseURL(env.GetSink())
 	if err != nil {
-		logger.Fatal("error building cloud event client", zap.Error(err))
+		logger.Fatal("error parsing sink URL", zap.Error(err))
+	}
+	sink := duckv1.Addressable{
+		URL:     sinkURL,
+		CACerts: env.GetCACerts(),
 	}
 
 	// Configuring the adapter
-	adapter := ctor(ctx, env, httpBindingsSender, reporter)
+	adapter := ctor(ctx, env, sink, reporter)
 
 	// Finally start the adapter (blocking)
 	if err := adapter.Start(ctx); err != nil {
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/eventtype_conversion.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/eventtype_conversion.go
index 663c82335..89911f65e 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/eventtype_conversion.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/eventtype_conversion.go
@@ -19,6 +19,8 @@ package v1beta1
 import (
 	"context"
 
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+
 	"knative.dev/eventing/pkg/apis/eventing/v1beta2"
 
 	"knative.dev/pkg/apis"
@@ -33,11 +35,15 @@ func (source *EventType) ConvertTo(ctx context.Context, obj apis.Convertible) er
 			Status: source.Status.Status,
 		}
 		sink.Spec = v1beta2.EventTypeSpec{
-			Type:        source.Spec.Type,
-			Source:      source.Spec.Source,
-			Schema:      source.Spec.Schema,
-			SchemaData:  source.Spec.SchemaData,
-			Broker:      source.Spec.Broker,
+			Type:       source.Spec.Type,
+			Source:     source.Spec.Source,
+			Schema:     source.Spec.Schema,
+			SchemaData: source.Spec.SchemaData,
+			Reference: &duckv1.KReference{
+				APIVersion: "eventing.knative.dev/v1",
+				Kind:       "Broker",
+				Name:       source.Spec.Broker,
+			},
 			Description: source.Spec.Description,
 		}
 
@@ -61,7 +67,7 @@ func (sink *EventType) ConvertFrom(ctx context.Context, obj apis.Convertible) er
 			Source:      source.Spec.Source,
 			Schema:      source.Spec.Schema,
 			SchemaData:  source.Spec.SchemaData,
-			Broker:      source.Spec.Broker,
+			Broker:      source.Spec.Reference.Name,
 			Description: source.Spec.Description,
 		}
 
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/eventtype_types.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/eventtype_types.go
index 648be1c08..c674e16fc 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/eventtype_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta1/eventtype_types.go
@@ -41,7 +41,6 @@ type EventType struct {
 	// Status represents the current state of the EventType.
 	// This data may be out of date.
 	// +optional
-	// TODO might be removed https://github.com/knative/eventing/issues/2750
 	Status EventTypeStatus `json:"status,omitempty"`
 }
 
@@ -77,7 +76,6 @@ type EventTypeSpec struct {
 	// The contents are not validated or manipulated by the system.
 	// +optional
 	SchemaData string `json:"schemaData,omitempty"`
-	// TODO remove https://github.com/knative/eventing/issues/2750
 	// Broker refers to the Broker that can provide the EventType.
 	// +optional
 	Broker string `json:"broker,omitempty"`
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta2/eventtype_defaults.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta2/eventtype_defaults.go
index 67ae41c88..3657729b5 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta2/eventtype_defaults.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta2/eventtype_defaults.go
@@ -16,14 +16,22 @@ limitations under the License.
 
 package v1beta2
 
-import "context"
+import (
+	"context"
+
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
 
 func (et *EventType) SetDefaults(ctx context.Context) {
 	et.Spec.SetDefaults(ctx)
 }
 
 func (ets *EventTypeSpec) SetDefaults(ctx context.Context) {
-	if ets.Broker == "" {
-		ets.Broker = "default"
+	if ets.Reference == nil {
+		ets.Reference = &duckv1.KReference{
+			APIVersion: "eventing.knative.dev/v1",
+			Kind:       "Broker",
+			Name:       "default",
+		}
 	}
 }
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta2/eventtype_types.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta2/eventtype_types.go
index e182e2d81..8356aa358 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta2/eventtype_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta2/eventtype_types.go
@@ -41,7 +41,6 @@ type EventType struct {
 	// Status represents the current state of the EventType.
 	// This data may be out of date.
 	// +optional
-	// TODO might be removed https://github.com/knative/eventing/issues/2750
 	Status EventTypeStatus `json:"status,omitempty"`
 }
 
@@ -77,10 +76,14 @@ type EventTypeSpec struct {
 	// The contents are not validated or manipulated by the system.
 	// +optional
 	SchemaData string `json:"schemaData,omitempty"`
-	// TODO remove https://github.com/knative/eventing/issues/2750
 	// Broker refers to the Broker that can provide the EventType.
+	// Deprecated: This field is deprecated and will be removed in a future release.
 	// +optional
 	Broker string `json:"broker,omitempty"`
+	// Reference is a KReference to the belonging addressable.
+	//For example, this could be a pointer to a Broker.
+	// +optional
+	Reference *duckv1.KReference `json:"reference,omitempty"`
 	// Description is an optional field used to describe the EventType, in any meaningful way.
 	// +optional
 	Description string `json:"description,omitempty"`
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta2/zz_generated.deepcopy.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta2/zz_generated.deepcopy.go
index 3163151d7..c3e2eb977 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta2/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1beta2/zz_generated.deepcopy.go
@@ -24,6 +24,7 @@ package v1beta2
 import (
 	runtime "k8s.io/apimachinery/pkg/runtime"
 	apis "knative.dev/pkg/apis"
+	v1 "knative.dev/pkg/apis/duck/v1"
 )
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
@@ -100,6 +101,11 @@ func (in *EventTypeSpec) DeepCopyInto(out *EventTypeSpec) {
 		*out = new(apis.URL)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.Reference != nil {
+		in, out := &in.Reference, &out.Reference
+		*out = new(v1.KReference)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
diff --git a/vendor/knative.dev/eventing/pkg/eventingtls/eventingtls.go b/vendor/knative.dev/eventing/pkg/eventingtls/eventingtls.go
index b787432aa..1207cbf2e 100644
--- a/vendor/knative.dev/eventing/pkg/eventingtls/eventingtls.go
+++ b/vendor/knative.dev/eventing/pkg/eventingtls/eventingtls.go
@@ -120,16 +120,16 @@ func GetCertificateFromSecret(ctx context.Context, informer coreinformersv1.Secr
 		},
 	})
 
-	// Store the current value so that we have certHolder initialized.
+	// If the Secret already exists, store its value
 	firstValue, err := informer.Lister().Secrets(secret.Namespace).Get(secret.Name)
 	if err != nil {
 		// Try to get the secret from the API Server when the lister failed.
-		firstValue, err = kube.CoreV1().Secrets(secret.Namespace).Get(ctx, secret.Name, metav1.GetOptions{})
-		if err != nil {
-			logger.Fatal(err.Error())
-		}
+		// Ignore any errors as the Secret may not be available yet.
+		firstValue, _ = kube.CoreV1().Secrets(secret.Namespace).Get(ctx, secret.Name, metav1.GetOptions{})
+	}
+	if firstValue != nil {
+		store(firstValue)
 	}
-	store(firstValue)
 
 	return func(info *tls.ClientHelloInfo) (*tls.Certificate, error) {
 		cert := certHolder.Load()
diff --git a/vendor/knative.dev/eventing/pkg/eventingtls/servermanager.go b/vendor/knative.dev/eventing/pkg/eventingtls/servermanager.go
index c1394b26a..10839ca92 100644
--- a/vendor/knative.dev/eventing/pkg/eventingtls/servermanager.go
+++ b/vendor/knative.dev/eventing/pkg/eventingtls/servermanager.go
@@ -22,7 +22,6 @@ import (
 	"net/http"
 
 	"knative.dev/eventing/pkg/apis/feature"
-	"knative.dev/eventing/pkg/kncloudevents"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/logging"
 )
@@ -35,14 +34,18 @@ import (
 // permissive: both http and https servers
 // strict: only https server
 type ServerManager struct {
-	httpReceiver  *kncloudevents.HTTPMessageReceiver
-	httpsReceiver *kncloudevents.HTTPMessageReceiver
+	httpReceiver  Receiver
+	httpsReceiver Receiver
 	handler       http.Handler
 	cmw           configmap.Watcher
 	featureStore  *feature.Store
 }
 
-func NewServerManager(ctx context.Context, httpReceiver, httpsReceiver *kncloudevents.HTTPMessageReceiver, handler http.Handler, cmw configmap.Watcher) (*ServerManager, error) {
+type Receiver interface {
+	StartListen(context.Context, http.Handler) error
+}
+
+func NewServerManager(ctx context.Context, httpReceiver, httpsReceiver Receiver, handler http.Handler, cmw configmap.Watcher) (*ServerManager, error) {
 	if httpReceiver == nil || httpsReceiver == nil {
 		return nil, fmt.Errorf("message receiver not provided")
 	}
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/http_client.go b/vendor/knative.dev/eventing/pkg/kncloudevents/http_client.go
index d9d938f73..c36d0afc0 100644
--- a/vendor/knative.dev/eventing/pkg/kncloudevents/http_client.go
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/http_client.go
@@ -19,35 +19,29 @@ package kncloudevents
 import (
 	nethttp "net/http"
 	"sync"
-	"time"
 
 	"go.opencensus.io/plugin/ochttp"
 	"knative.dev/pkg/tracing/propagation/tracecontextb3"
 )
 
-const (
-	defaultRetryWaitMin = 1 * time.Second
-	defaultRetryWaitMax = 30 * time.Second
-)
-
-type holder struct {
+type legacyHolder struct {
 	clientMutex    sync.Mutex
 	connectionArgs *ConnectionArgs
 	client         **nethttp.Client
 }
 
-var clientHolder = holder{}
+var legacyClientHolder = legacyHolder{}
 
 // The used HTTP client is a singleton, so the same http client is reused across all the application.
 // If connection args is modified, client is cleaned and a new one is created.
 func getClient() *nethttp.Client {
-	clientHolder.clientMutex.Lock()
-	defer clientHolder.clientMutex.Unlock()
+	legacyClientHolder.clientMutex.Lock()
+	defer legacyClientHolder.clientMutex.Unlock()
 
-	if clientHolder.client == nil {
+	if legacyClientHolder.client == nil {
 		// Add connection options to the default transport.
 		var base = nethttp.DefaultTransport.(*nethttp.Transport).Clone()
-		clientHolder.connectionArgs.configureTransport(base)
+		legacyClientHolder.connectionArgs.configureTransport(base)
 		c := &nethttp.Client{
 			// Add output tracing.
 			Transport: &ochttp.Transport{
@@ -55,52 +49,35 @@ func getClient() *nethttp.Client {
 				Propagation: tracecontextb3.TraceContextEgress,
 			},
 		}
-		clientHolder.client = &c
+		legacyClientHolder.client = &c
 	}
 
-	return *clientHolder.client
+	return *legacyClientHolder.client
 }
 
 // ConfigureConnectionArgs configures the new connection args.
 // The existing client won't be affected, but a new one will be created.
 // Use sparingly, because it might lead to creating a lot of clients, none of them sharing their connection pool!
-func ConfigureConnectionArgs(ca *ConnectionArgs) {
-	clientHolder.clientMutex.Lock()
-	defer clientHolder.clientMutex.Unlock()
+func configureConnectionArgsOldClient(ca *ConnectionArgs) {
+	legacyClientHolder.clientMutex.Lock()
+	defer legacyClientHolder.clientMutex.Unlock()
 
 	// Check if same config
-	if clientHolder.connectionArgs != nil &&
+	if legacyClientHolder.connectionArgs != nil &&
 		ca != nil &&
-		ca.MaxIdleConns == clientHolder.connectionArgs.MaxIdleConns &&
-		ca.MaxIdleConnsPerHost == clientHolder.connectionArgs.MaxIdleConnsPerHost {
+		ca.MaxIdleConns == legacyClientHolder.connectionArgs.MaxIdleConns &&
+		ca.MaxIdleConnsPerHost == legacyClientHolder.connectionArgs.MaxIdleConnsPerHost {
 		return
 	}
 
-	if clientHolder.client != nil {
+	if legacyClientHolder.client != nil {
 		// Let's try to clean up a bit the existing client
 		// Note: this won't remove it nor close it
-		(*clientHolder.client).CloseIdleConnections()
+		(*legacyClientHolder.client).CloseIdleConnections()
 
 		// Setting client to nil
-		clientHolder.client = nil
+		legacyClientHolder.client = nil
 	}
 
-	clientHolder.connectionArgs = ca
-}
-
-// ConnectionArgs allow to configure connection parameters to the underlying
-// HTTP Client transport.
-type ConnectionArgs struct {
-	// MaxIdleConns refers to the max idle connections, as in net/http/transport.
-	MaxIdleConns int
-	// MaxIdleConnsPerHost refers to the max idle connections per host, as in net/http/transport.
-	MaxIdleConnsPerHost int
-}
-
-func (ca *ConnectionArgs) configureTransport(transport *nethttp.Transport) {
-	if ca == nil {
-		return
-	}
-	transport.MaxIdleConns = ca.MaxIdleConns
-	transport.MaxIdleConnsPerHost = ca.MaxIdleConnsPerHost
+	legacyClientHolder.connectionArgs = ca
 }
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/http_client_new.go b/vendor/knative.dev/eventing/pkg/kncloudevents/http_client_new.go
new file mode 100644
index 000000000..0c9f1be1a
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/http_client_new.go
@@ -0,0 +1,168 @@
+/*
+Copyright 2023 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 kncloudevents
+
+import (
+	"fmt"
+	nethttp "net/http"
+	"sync"
+	"time"
+
+	"go.opencensus.io/plugin/ochttp"
+	"knative.dev/eventing/pkg/eventingtls"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/tracing/propagation/tracecontextb3"
+)
+
+const (
+	defaultRetryWaitMin = 1 * time.Second
+	defaultRetryWaitMax = 30 * time.Second
+)
+
+var (
+	clients clientsHolder
+)
+
+type clientsHolder struct {
+	mu             sync.Mutex
+	clients        map[string]*nethttp.Client
+	connectionArgs *ConnectionArgs
+}
+
+func init() {
+	clients = clientsHolder{
+		clients: make(map[string]*nethttp.Client),
+	}
+}
+
+func getClientForAddressable(addressable duckv1.Addressable) (*nethttp.Client, error) {
+	clients.mu.Lock()
+	defer clients.mu.Unlock()
+
+	clientKey := addressable.URL.String()
+
+	client, ok := clients.clients[clientKey]
+	if !ok {
+		newClient, err := createNewClient(addressable)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create new client for addressable: %w", err)
+		}
+
+		clients.clients[clientKey] = newClient
+
+		client = newClient
+	}
+
+	return client, nil
+}
+
+func createNewClient(addressable duckv1.Addressable) (*nethttp.Client, error) {
+	var base = nethttp.DefaultTransport.(*nethttp.Transport).Clone()
+
+	if addressable.CACerts != nil && *addressable.CACerts != "" {
+		var err error
+
+		clientConfig := eventingtls.NewDefaultClientConfig()
+		clientConfig.CACerts = addressable.CACerts
+
+		base.TLSClientConfig, err = eventingtls.GetTLSClientConfig(clientConfig)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	clients.connectionArgs.configureTransport(base)
+	client := &nethttp.Client{
+		// Add output tracing.
+		Transport: &ochttp.Transport{
+			Base:        base,
+			Propagation: tracecontextb3.TraceContextEgress,
+		},
+	}
+
+	return client, nil
+}
+
+func AddOrUpdateAddressableHandler(addressable duckv1.Addressable) {
+	clients.mu.Lock()
+	defer clients.mu.Unlock()
+
+	clientKey := addressable.URL.String()
+
+	client, err := createNewClient(addressable)
+	if err != nil {
+		fmt.Printf("failed to create new client: %v", err)
+		return
+	}
+	clients.clients[clientKey] = client
+}
+
+func DeleteAddressableHandler(addressable duckv1.Addressable) {
+	clients.mu.Lock()
+	defer clients.mu.Unlock()
+
+	clientKey := addressable.URL.String()
+
+	delete(clients.clients, clientKey)
+}
+
+// ConfigureConnectionArgs configures the new connection args.
+// Use sparingly, because it might lead to creating a lot of clients, none of them sharing their connection pool!
+func ConfigureConnectionArgs(ca *ConnectionArgs) {
+	configureConnectionArgsOldClient(ca) //also configure the connection args of the old client
+
+	clients.mu.Lock()
+	defer clients.mu.Unlock()
+
+	// Check if same config
+	if clients.connectionArgs != nil &&
+		ca != nil &&
+		ca.MaxIdleConns == clients.connectionArgs.MaxIdleConns &&
+		ca.MaxIdleConnsPerHost == clients.connectionArgs.MaxIdleConnsPerHost {
+		return
+	}
+
+	if len(clients.clients) > 0 {
+		// Let's try to clean up a bit the existing clients
+		// Note: this won't remove it nor close it
+		for _, client := range clients.clients {
+			client.CloseIdleConnections()
+		}
+
+		// Resetting clients
+		clients.clients = make(map[string]*nethttp.Client)
+	}
+
+	clients.connectionArgs = ca
+}
+
+// ConnectionArgs allow to configure connection parameters to the underlying
+// HTTP Client transport.
+type ConnectionArgs struct {
+	// MaxIdleConns refers to the max idle connections, as in net/http/transport.
+	MaxIdleConns int
+	// MaxIdleConnsPerHost refers to the max idle connections per host, as in net/http/transport.
+	MaxIdleConnsPerHost int
+}
+
+func (ca *ConnectionArgs) configureTransport(transport *nethttp.Transport) {
+	if ca == nil {
+		return
+	}
+	transport.MaxIdleConns = ca.MaxIdleConns
+	transport.MaxIdleConnsPerHost = ca.MaxIdleConnsPerHost
+}
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
index 1940439ad..b026bbf43 100644
--- a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
@@ -18,17 +18,13 @@ package kncloudevents
 
 import (
 	"context"
-	"fmt"
 	nethttp "net/http"
-	"strconv"
-	"time"
 
 	"github.com/hashicorp/go-retryablehttp"
 )
 
-const RetryAfterHeader = "Retry-After"
-
-// HTTPMessageSender is a wrapper for an http client that can send cloudevents.Request with retries
+// HTTPMessageSender is a wrapper for an http client that can send cloudevents.Request with retries.
+// Deprecated: Use kncloudevents.CloudEventsRequest instead.
 type HTTPMessageSender struct {
 	Client *nethttp.Client
 	Target string
@@ -84,91 +80,3 @@ func (s *HTTPMessageSender) SendWithRetries(req *nethttp.Request, config *RetryC
 
 	return retryableClient.Do(retryableReq)
 }
-
-// generateBackoffFunction returns a valid retryablehttp.Backoff implementation which
-// wraps the provided RetryConfig.Backoff implementation with optional "Retry-After"
-// header support.
-func generateBackoffFn(config *RetryConfig) retryablehttp.Backoff {
-	return func(_, _ time.Duration, attemptNum int, resp *nethttp.Response) time.Duration {
-
-		//
-		// NOTE - The following logic will need to be altered slightly once the "delivery-retryafter"
-		//        experimental-feature graduates from Alpha/Beta to Stable/GA.  This is according to
-		//        plan as described in https://github.com/knative/eventing/issues/5811.
-		//
-		//        During the Alpha/Beta stages the ability to respect Retry-After headers is "opt-in"
-		//        requiring the DeliverySpec.RetryAfterMax to be populated.  The Stable/GA behavior
-		//        will be "opt-out" where Retry-After headers are always respected (in the context of
-		//        calculating backoff durations for 429 / 503 responses) unless the
-		//        DeliverySpec.RetryAfterMax is set to "PT0S".
-		//
-		//        While this might seem unnecessarily complex, it achieves the following design goals...
-		//          - Does not require an explicit "enabled" flag in the DeliverySpec.
-		//          - Does not require implementations calling the message_sender to be aware of experimental-features.
-		//          - Does not modify existing Knative CRs with arbitrary default "max" values.
-		//
-		//        The intended behavior of RetryConfig.RetryAfterMaxDuration is as follows...
-		//
-		//          RetryAfterMaxDuration    Alpha/Beta                              Stable/GA
-		//          ---------------------    ----------                              ---------
-		//               nil                 Do NOT respect Retry-After headers      Respect Retry-After headers without Max
-		//                0                  Do NOT respect Retry-After headers      Do NOT respect Retry-After headers
-		//               >0                  Respect Retry-After headers with Max    Respect Retry-After headers with Max
-		//
-
-		// If Response is 429 / 503, Then Parse Any Retry-After Header Durations & Enforce Optional MaxDuration
-		var retryAfterDuration time.Duration
-		// TODO - Remove this check when experimental-feature moves to Stable/GA to convert behavior from opt-in to opt-out
-		if config.RetryAfterMaxDuration != nil {
-			// TODO - Keep this logic as is (no change required) when experimental-feature is Stable/GA
-			if resp != nil && (resp.StatusCode == nethttp.StatusTooManyRequests || resp.StatusCode == nethttp.StatusServiceUnavailable) {
-				retryAfterDuration = parseRetryAfterDuration(resp)
-				if config.RetryAfterMaxDuration != nil && *config.RetryAfterMaxDuration < retryAfterDuration {
-					retryAfterDuration = *config.RetryAfterMaxDuration
-				}
-			}
-		}
-
-		// Calculate The RetryConfig Backoff Duration
-		backoffDuration := config.Backoff(attemptNum, resp)
-
-		// Return The Larger Of The Two Backoff Durations
-		if retryAfterDuration > backoffDuration {
-			return retryAfterDuration
-		}
-		return backoffDuration
-	}
-}
-
-// parseRetryAfterDuration returns a Duration expressing the amount of time
-// requested to wait by a Retry-After header, or 0 if not present or invalid.
-// According to the spec (https://tools.ietf.org/html/rfc7231#section-7.1.3)
-// the Retry-After Header's value can be one of an HTTP-date or delay-seconds,
-// both of which are supported here.
-func parseRetryAfterDuration(resp *nethttp.Response) (retryAfterDuration time.Duration) {
-
-	// Return 0 Duration If No Response / Headers
-	if resp == nil || resp.Header == nil {
-		return
-	}
-
-	// Return 0 Duration If No Retry-After Header
-	retryAfterString := resp.Header.Get(RetryAfterHeader)
-	if len(retryAfterString) <= 0 {
-		return
-	}
-
-	// Attempt To Parse Retry-After Header As Seconds - Return If Successful
-	retryAfterInt, parseIntErr := strconv.ParseInt(retryAfterString, 10, 64)
-	if parseIntErr == nil {
-		return time.Duration(retryAfterInt) * time.Second
-	}
-
-	// Attempt To Parse Retry-After Header As Timestamp (RFC850 & ANSIC) - Return If Successful
-	retryAfterTime, parseTimeErr := nethttp.ParseTime(retryAfterString)
-	if parseTimeErr != nil {
-		fmt.Printf("failed to parse Retry-After header: ParseInt Error = %v, ParseTime Error = %v\n", parseIntErr, parseTimeErr)
-		return
-	}
-	return time.Until(retryAfterTime)
-}
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender_new.go b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender_new.go
new file mode 100644
index 000000000..16f3accd7
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender_new.go
@@ -0,0 +1,182 @@
+/*
+Copyright 2023 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 kncloudevents
+
+import (
+	"context"
+	"fmt"
+	nethttp "net/http"
+	"strconv"
+	"time"
+
+	"github.com/hashicorp/go-retryablehttp"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+const RetryAfterHeader = "Retry-After"
+
+type CloudEventRequest struct {
+	*nethttp.Request
+	Target duckv1.Addressable
+}
+
+func (req *CloudEventRequest) Send() (*nethttp.Response, error) {
+	client, err := getClientForAddressable(req.Target)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get client for addressable: %w", err)
+	}
+	return client.Do(req.Request)
+}
+
+func (req *CloudEventRequest) SendWithRetries(config *RetryConfig) (*nethttp.Response, error) {
+	if config == nil {
+		return req.Send()
+	}
+
+	client, err := getClientForAddressable(req.Target)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get client for addressable: %w", err)
+	}
+
+	if config.RequestTimeout != 0 {
+		client = &nethttp.Client{
+			Transport:     client.Transport,
+			CheckRedirect: client.CheckRedirect,
+			Jar:           client.Jar,
+			Timeout:       config.RequestTimeout,
+		}
+	}
+
+	retryableClient := retryablehttp.Client{
+		HTTPClient:   client,
+		RetryWaitMin: defaultRetryWaitMin,
+		RetryWaitMax: defaultRetryWaitMax,
+		RetryMax:     config.RetryMax,
+		CheckRetry:   retryablehttp.CheckRetry(config.CheckRetry),
+		Backoff:      generateBackoffFn(config),
+		ErrorHandler: func(resp *nethttp.Response, err error, numTries int) (*nethttp.Response, error) {
+			return resp, err
+		},
+	}
+
+	retryableReq, err := retryablehttp.FromRequest(req.Request)
+	if err != nil {
+		return nil, err
+	}
+
+	return retryableClient.Do(retryableReq)
+}
+
+func NewCloudEventRequest(ctx context.Context, target duckv1.Addressable) (*CloudEventRequest, error) {
+	nethttpReqest, err := nethttp.NewRequestWithContext(ctx, "POST", target.URL.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+
+	return &CloudEventRequest{
+		Target:  target,
+		Request: nethttpReqest,
+	}, nil
+}
+
+// generateBackoffFunction returns a valid retryablehttp.Backoff implementation which
+// wraps the provided RetryConfig.Backoff implementation with optional "Retry-After"
+// header support.
+func generateBackoffFn(config *RetryConfig) retryablehttp.Backoff {
+	return func(_, _ time.Duration, attemptNum int, resp *nethttp.Response) time.Duration {
+
+		//
+		// NOTE - The following logic will need to be altered slightly once the "delivery-retryafter"
+		//        experimental-feature graduates from Alpha/Beta to Stable/GA.  This is according to
+		//        plan as described in https://github.com/knative/eventing/issues/5811.
+		//
+		//        During the Alpha/Beta stages the ability to respect Retry-After headers is "opt-in"
+		//        requiring the DeliverySpec.RetryAfterMax to be populated.  The Stable/GA behavior
+		//        will be "opt-out" where Retry-After headers are always respected (in the context of
+		//        calculating backoff durations for 429 / 503 responses) unless the
+		//        DeliverySpec.RetryAfterMax is set to "PT0S".
+		//
+		//        While this might seem unnecessarily complex, it achieves the following design goals...
+		//          - Does not require an explicit "enabled" flag in the DeliverySpec.
+		//          - Does not require implementations calling the message_sender to be aware of experimental-features.
+		//          - Does not modify existing Knative CRs with arbitrary default "max" values.
+		//
+		//        The intended behavior of RetryConfig.RetryAfterMaxDuration is as follows...
+		//
+		//          RetryAfterMaxDuration    Alpha/Beta                              Stable/GA
+		//          ---------------------    ----------                              ---------
+		//               nil                 Do NOT respect Retry-After headers      Respect Retry-After headers without Max
+		//                0                  Do NOT respect Retry-After headers      Do NOT respect Retry-After headers
+		//               >0                  Respect Retry-After headers with Max    Respect Retry-After headers with Max
+		//
+
+		// If Response is 429 / 503, Then Parse Any Retry-After Header Durations & Enforce Optional MaxDuration
+		var retryAfterDuration time.Duration
+		// TODO - Remove this check when experimental-feature moves to Stable/GA to convert behavior from opt-in to opt-out
+		if config.RetryAfterMaxDuration != nil {
+			// TODO - Keep this logic as is (no change required) when experimental-feature is Stable/GA
+			if resp != nil && (resp.StatusCode == nethttp.StatusTooManyRequests || resp.StatusCode == nethttp.StatusServiceUnavailable) {
+				retryAfterDuration = parseRetryAfterDuration(resp)
+				if config.RetryAfterMaxDuration != nil && *config.RetryAfterMaxDuration < retryAfterDuration {
+					retryAfterDuration = *config.RetryAfterMaxDuration
+				}
+			}
+		}
+
+		// Calculate The RetryConfig Backoff Duration
+		backoffDuration := config.Backoff(attemptNum, resp)
+
+		// Return The Larger Of The Two Backoff Durations
+		if retryAfterDuration > backoffDuration {
+			return retryAfterDuration
+		}
+		return backoffDuration
+	}
+}
+
+// parseRetryAfterDuration returns a Duration expressing the amount of time
+// requested to wait by a Retry-After header, or 0 if not present or invalid.
+// According to the spec (https://tools.ietf.org/html/rfc7231#section-7.1.3)
+// the Retry-After Header's value can be one of an HTTP-date or delay-seconds,
+// both of which are supported here.
+func parseRetryAfterDuration(resp *nethttp.Response) (retryAfterDuration time.Duration) {
+
+	// Return 0 Duration If No Response / Headers
+	if resp == nil || resp.Header == nil {
+		return
+	}
+
+	// Return 0 Duration If No Retry-After Header
+	retryAfterString := resp.Header.Get(RetryAfterHeader)
+	if len(retryAfterString) <= 0 {
+		return
+	}
+
+	// Attempt To Parse Retry-After Header As Seconds - Return If Successful
+	retryAfterInt, parseIntErr := strconv.ParseInt(retryAfterString, 10, 64)
+	if parseIntErr == nil {
+		return time.Duration(retryAfterInt) * time.Second
+	}
+
+	// Attempt To Parse Retry-After Header As Timestamp (RFC850 & ANSIC) - Return If Successful
+	retryAfterTime, parseTimeErr := nethttp.ParseTime(retryAfterString)
+	if parseTimeErr != nil {
+		fmt.Printf("failed to parse Retry-After header: ParseInt Error = %v, ParseTime Error = %v\n", parseIntErr, parseTimeErr)
+		return
+	}
+	return time.Until(retryAfterTime)
+}
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/utils.go b/vendor/knative.dev/eventing/pkg/kncloudevents/utils.go
index 7649adf35..02e58497b 100644
--- a/vendor/knative.dev/eventing/pkg/kncloudevents/utils.go
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/utils.go
@@ -40,6 +40,11 @@ func WriteHTTPRequestWithAdditionalHeaders(ctx context.Context, message binding.
 	return nil
 }
 
+func WriteRequestWithAdditionalHeaders(ctx context.Context, message binding.Message, req *CloudEventRequest,
+	additionalHeaders nethttp.Header, transformers ...binding.Transformer) error {
+	return WriteHTTPRequestWithAdditionalHeaders(ctx, message, req.Request, additionalHeaders, transformers...)
+}
+
 type TypeExtractorTransformer string
 
 func (a *TypeExtractorTransformer) Transform(reader binding.MessageMetadataReader, _ binding.MessageMetadataWriter) error {
diff --git a/vendor/knative.dev/pkg/injection/clients.go b/vendor/knative.dev/pkg/injection/clients.go
index 92e991218..b71ef1d94 100644
--- a/vendor/knative.dev/pkg/injection/clients.go
+++ b/vendor/knative.dev/pkg/injection/clients.go
@@ -62,22 +62,3 @@ func (i *impl) FetchAllClients(ctx context.Context) []interface{} {
 	}
 	return clients
 }
-
-// DynamicClientInjector holds the type of a callback that attaches a particular
-// client type to a context.
-type DynamicClientInjector func(context.Context) context.Context
-
-func (i *impl) RegisterDynamicClient(ci DynamicClientInjector) {
-	i.m.Lock()
-	defer i.m.Unlock()
-
-	i.dynamicClients = append(i.dynamicClients, ci)
-}
-
-func (i *impl) GetDynamicClients() []DynamicClientInjector {
-	i.m.RLock()
-	defer i.m.RUnlock()
-
-	// Copy the slice before returning.
-	return append(i.dynamicClients[:0:0], i.dynamicClients...)
-}
diff --git a/vendor/knative.dev/pkg/injection/informers.go b/vendor/knative.dev/pkg/injection/informers.go
index ce5d481e8..9356f8d7f 100644
--- a/vendor/knative.dev/pkg/injection/informers.go
+++ b/vendor/knative.dev/pkg/injection/informers.go
@@ -28,10 +28,6 @@ import (
 // informer type to a context.
 type InformerInjector func(context.Context) (context.Context, controller.Informer)
 
-// DynamicInformerInjector holds the type of a callback that attaches a particular
-// informer type (backed by a Dynamic) to a context.
-type DynamicInformerInjector func(context.Context) context.Context
-
 // FilteredInformersInjector holds the type of a callback that attaches a set of particular
 // filtered informers type to a context.
 type FilteredInformersInjector func(context.Context) (context.Context, []controller.Informer)
@@ -43,13 +39,6 @@ func (i *impl) RegisterInformer(ii InformerInjector) {
 	i.informers = append(i.informers, ii)
 }
 
-func (i *impl) RegisterDynamicInformer(ii DynamicInformerInjector) {
-	i.m.Lock()
-	defer i.m.Unlock()
-
-	i.dynamicInformers = append(i.dynamicInformers, ii)
-}
-
 func (i *impl) RegisterFilteredInformers(fii FilteredInformersInjector) {
 	i.m.Lock()
 	defer i.m.Unlock()
@@ -65,14 +54,6 @@ func (i *impl) GetInformers() []InformerInjector {
 	return append(i.informers[:0:0], i.informers...)
 }
 
-func (i *impl) GetDynamicInformers() []DynamicInformerInjector {
-	i.m.RLock()
-	defer i.m.RUnlock()
-
-	// Copy the slice before returning.
-	return append(i.dynamicInformers[:0:0], i.dynamicInformers...)
-}
-
 func (i *impl) GetFilteredInformers() []FilteredInformersInjector {
 	i.m.RLock()
 	defer i.m.RUnlock()
@@ -81,22 +62,6 @@ func (i *impl) GetFilteredInformers() []FilteredInformersInjector {
 	return append(i.filteredInformers[:0:0], i.filteredInformers...)
 }
 
-func (i *impl) SetupDynamic(ctx context.Context) context.Context {
-	// Based on the reconcilers we have linked, build up a set of clients and inject
-	// them onto the context.
-	for _, ci := range i.GetDynamicClients() {
-		ctx = ci(ctx)
-	}
-
-	// Based on the reconcilers we have linked, build up a set of informers
-	// and inject them onto the context.
-	for _, ii := range i.GetDynamicInformers() {
-		ctx = ii(ctx)
-	}
-
-	return ctx
-}
-
 func (i *impl) SetupInformers(ctx context.Context, cfg *rest.Config) (context.Context, []controller.Informer) {
 	// Based on the reconcilers we have linked, build up a set of clients and inject
 	// them onto the context.
diff --git a/vendor/knative.dev/pkg/injection/interface.go b/vendor/knative.dev/pkg/injection/interface.go
index 158864015..c6d5715ad 100644
--- a/vendor/knative.dev/pkg/injection/interface.go
+++ b/vendor/knative.dev/pkg/injection/interface.go
@@ -78,29 +78,6 @@ type Interface interface {
 	SetupInformers(context.Context, *rest.Config) (context.Context, []controller.Informer)
 }
 
-// DynamicInterface is the interface for interacting with dynamicclient-based injection
-// implementations, such as Dynamic below.
-type DynamicInterface interface {
-	// RegisterDynamicClient registers a new injector callback for associating
-	// a new dynamicclient-based client with a context.
-	RegisterDynamicClient(DynamicClientInjector)
-
-	// GetDynamicClients fetches all of the registered dynamicclient-based client injectors.
-	GetDynamicClients() []DynamicClientInjector
-
-	// RegisterDynamicInformer registers a new injector callback for associating
-	// a new dynamicclient-based informer with a context.
-	RegisterDynamicInformer(DynamicInformerInjector)
-
-	// GetDynamicInformers fetches all of the registered dynamicclient-based informer injectors.
-	GetDynamicInformers() []DynamicInformerInjector
-
-	// SetupDynamic runs all of the injectors against a context, starting with
-	// the clients and the given stream.  A context infused with the various elements
-	// is returned.
-	SetupDynamic(context.Context) context.Context
-}
-
 type ControllerConstructor func(context.Context, configmap.Watcher) *controller.Impl
 
 // NamedControllerConstructor is a ControllerConstructor with an associated name.
@@ -120,10 +97,6 @@ var (
 	// are being run for real.
 	Default Interface = &impl{}
 
-	// Dynamic is the injection interface to use when bootstrapping a version
-	// of things based on the prototype dynamicclient-based reconciler framework.
-	Dynamic DynamicInterface = &impl{}
-
 	// Fake is the injection interface with which informers should register
 	// to make themselves available to the controller process when it is being
 	// unit tested.
@@ -134,11 +107,9 @@ type impl struct {
 	m sync.RWMutex
 
 	clients           []ClientInjector
-	dynamicClients    []DynamicClientInjector
 	clientFetchers    []ClientFetcher
 	factories         []InformerFactoryInjector
 	informers         []InformerInjector
-	dynamicInformers  []DynamicInformerInjector
 	filteredInformers []FilteredInformersInjector
 	ducks             []DuckFactoryInjector
 }
diff --git a/vendor/knative.dev/pkg/webhook/webhook.go b/vendor/knative.dev/pkg/webhook/webhook.go
index 779d388d2..80bd8c2be 100644
--- a/vendor/knative.dev/pkg/webhook/webhook.go
+++ b/vendor/knative.dev/pkg/webhook/webhook.go
@@ -22,6 +22,8 @@ import (
 	"errors"
 	"fmt"
 	"html"
+	"log"
+	"net"
 	"net/http"
 	"time"
 
@@ -94,6 +96,9 @@ type Webhook struct {
 
 	// The TLS configuration to use for serving (or nil for non-TLS)
 	tlsConfig *tls.Config
+
+	// testListener is only used in testing so we don't get port conflicts
+	testListener net.Listener
 }
 
 // New constructs a Webhook
@@ -196,7 +201,6 @@ func New(
 		default:
 			return nil, fmt.Errorf("unknown webhook controller type:  %T", controller)
 		}
-
 	}
 
 	return
@@ -209,6 +213,15 @@ func (wh *Webhook) InformersHaveSynced() {
 	wh.Logger.Info("Informers have been synced, unblocking admission webhooks.")
 }
 
+type zapWrapper struct {
+	logger *zap.SugaredLogger
+}
+
+func (z *zapWrapper) Write(p []byte) (n int, err error) {
+	z.logger.Errorw(string(p))
+	return len(p), nil
+}
+
 // Run implements the admission controller run loop.
 func (wh *Webhook) Run(stop <-chan struct{}) error {
 	logger := wh.Logger
@@ -220,24 +233,34 @@ func (wh *Webhook) Run(stop <-chan struct{}) error {
 	}
 
 	server := &http.Server{
+		ErrorLog:          log.New(&zapWrapper{logger}, "", 0),
 		Handler:           drainer,
 		Addr:              fmt.Sprint(":", wh.Options.Port),
 		TLSConfig:         wh.tlsConfig,
 		ReadHeaderTimeout: time.Minute, //https://medium.com/a-journey-with-go/go-understand-and-mitigate-slowloris-attack-711c1b1403f6
 	}
 
+	var serve = server.ListenAndServe
+
+	if server.TLSConfig != nil && wh.testListener != nil {
+		serve = func() error {
+			return server.ServeTLS(wh.testListener, "", "")
+		}
+	} else if server.TLSConfig != nil {
+		serve = func() error {
+			return server.ListenAndServeTLS("", "")
+		}
+	} else if wh.testListener != nil {
+		serve = func() error {
+			return server.Serve(wh.testListener)
+		}
+	}
+
 	eg, ctx := errgroup.WithContext(ctx)
 	eg.Go(func() error {
-		if server.TLSConfig != nil {
-			if err := server.ListenAndServeTLS("", ""); err != nil && !errors.Is(err, http.ErrServerClosed) {
-				logger.Errorw("ListenAndServeTLS for admission webhook returned error", zap.Error(err))
-				return err
-			}
-		} else {
-			if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
-				logger.Errorw("ListenAndServe for admission webhook returned error", zap.Error(err))
-				return err
-			}
+		if err := serve(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			logger.Errorw("ListenAndServe for admission webhook returned error", zap.Error(err))
+			return err
 		}
 		return nil
 	})
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 4331ae74c..f28b8a241 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1307,7 +1307,7 @@ k8s.io/utils/net
 k8s.io/utils/pointer
 k8s.io/utils/strings/slices
 k8s.io/utils/trace
-# knative.dev/eventing v0.37.1-0.20230524182623-6a11c5f02ccf
+# knative.dev/eventing v0.37.1-0.20230531153536-70113e887eec
 ## explicit; go 1.19
 knative.dev/eventing/cmd/heartbeats
 knative.dev/eventing/pkg/adapter/v2
@@ -1413,11 +1413,11 @@ knative.dev/eventing/test/upgrade/prober/wathola/fetcher
 knative.dev/eventing/test/upgrade/prober/wathola/forwarder
 knative.dev/eventing/test/upgrade/prober/wathola/receiver
 knative.dev/eventing/test/upgrade/prober/wathola/sender
-# knative.dev/hack v0.0.0-20230524013611-5812c57cc1ac
+# knative.dev/hack v0.0.0-20230530201435-5b7907fbe350
 ## explicit; go 1.18
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/pkg v0.0.0-20230524013411-ec20442c6ead
+# knative.dev/pkg v0.0.0-20230531073936-5671699f23d9
 ## explicit; go 1.18
 knative.dev/pkg/apiextensions/storageversion
 knative.dev/pkg/apiextensions/storageversion/cmd/migrate
@@ -1515,7 +1515,7 @@ knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/conversion
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20230524170754-968cbdf0912e
+# knative.dev/reconciler-test v0.0.0-20230530092335-cf14f44032b9
 ## explicit; go 1.18
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/pkg/source/adapter/adapter_benchmark_test.go b/pkg/source/adapter/adapter_benchmark_test.go
deleted file mode 100644
index 1393f9b9d..000000000
--- a/pkg/source/adapter/adapter_benchmark_test.go
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
-Copyright 2019 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 kafka
-
-import (
-	"bytes"
-	"context"
-	"encoding/json"
-	"io"
-	"net/http"
-	"strconv"
-	"testing"
-	"time"
-
-	"knative.dev/eventing/pkg/adapter/v2"
-	"knative.dev/eventing/pkg/kncloudevents"
-	"knative.dev/eventing/pkg/metrics/source"
-
-	"github.com/Shopify/sarama"
-	"go.uber.org/zap"
-)
-
-// Run with go test -v ./kafka/source/pkg/adapter/ -gcflags="-N -l" -test.benchtime 2s -benchmem -run=Handle -bench=.
-
-type RoundTripFunc func(req *http.Request) *http.Response
-
-func (f RoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
-	return f(req), nil
-}
-
-func NewTestClient(fn RoundTripFunc) *http.Client {
-	return &http.Client{
-		Transport: fn,
-	}
-}
-
-func BenchmarkHandle(b *testing.B) {
-	sinkUrl := "http://localhost:8080" // This address doesn't matter, it's not used b/c we mock the client transport
-
-	data, err := json.Marshal(map[string]string{"key": "value"})
-	if err != nil {
-		b.Errorf("unexpected error, %v", err)
-	}
-
-	s := kncloudevents.HTTPMessageSender{
-		Client: NewTestClient(func(req *http.Request) *http.Response {
-			return &http.Response{
-				StatusCode: 202,
-				Header:     make(http.Header),
-				Body:       io.NopCloser(bytes.NewReader([]byte{})),
-			}
-		}),
-		Target: sinkUrl,
-	}
-	if err != nil {
-		b.Fatal(err)
-	}
-
-	statsReporter, _ := source.NewStatsReporter()
-
-	a := &Adapter{
-		config: &AdapterConfig{
-			EnvConfig: adapter.EnvConfig{
-				Sink:      sinkUrl,
-				Namespace: "test",
-			},
-			Topics:        []string{"topic1", "topic2"},
-			ConsumerGroup: "group",
-			Name:          "test",
-		},
-		httpMessageSender: &s,
-		logger:            zap.NewNop().Sugar(),
-		keyTypeMapper:     getKeyTypeMapper(""),
-		reporter:          statsReporter,
-	}
-	b.SetParallelism(1)
-	b.Run("Baseline", func(b *testing.B) {
-		baseline(b, a, data)
-	})
-	b.Run("Benchmark Handle", func(b *testing.B) {
-		benchmarkHandle(b, a, data)
-	})
-
-}
-
-// Avoid DCE
-var Message *sarama.ConsumerMessage
-var ABool bool
-var AError error
-
-// Baseline is required to understand how much time/memory is needed to allocate the sarama.ConsumerMessage
-func baseline(b *testing.B, adapter *Adapter, payload []byte) {
-	for i := 0; i < b.N; i++ {
-		Message = &sarama.ConsumerMessage{
-			Key:       []byte(strconv.Itoa(i)),
-			Topic:     "topic1",
-			Value:     payload,
-			Partition: 1,
-			Offset:    int64(i + 1),
-			Timestamp: time.Now(),
-		}
-	}
-}
-
-func benchmarkHandle(b *testing.B, adapter *Adapter, payload []byte) {
-	for i := 0; i < b.N; i++ {
-		Message := &sarama.ConsumerMessage{
-			Key:       []byte(strconv.Itoa(i)),
-			Topic:     "topic1",
-			Value:     payload,
-			Partition: 1,
-			Offset:    int64(i + 1),
-			Timestamp: time.Now(),
-		}
-		ABool, AError = adapter.Handle(context.TODO(), Message)
-		if AError != nil {
-			panic(AError)
-		}
-	}
-}
diff --git a/pkg/source/adapter/adapter_test.go b/pkg/source/adapter/adapter_test.go
index 9895829a2..92d90437b 100644
--- a/pkg/source/adapter/adapter_test.go
+++ b/pkg/source/adapter/adapter_test.go
@@ -30,8 +30,9 @@ import (
 	"github.com/cloudevents/sdk-go/v2/types"
 	"go.uber.org/zap"
 	"knative.dev/eventing/pkg/adapter/v2"
-	"knative.dev/eventing/pkg/kncloudevents"
 	"knative.dev/eventing/pkg/metrics/source"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 )
@@ -333,9 +334,12 @@ func TestPostMessage_ServeHTTP_binary_mode(t *testing.T) {
 			//})
 			//defer time.Sleep(1 * time.Second)
 
-			s, err := kncloudevents.NewHTTPMessageSenderWithTarget(sinkServer.URL)
+			sinkURI, err := apis.ParseURL(sinkServer.URL)
 			if err != nil {
-				t.Fatal(err)
+				t.Fatalf("Unexpected error parsing sinkURL: %v", err)
+			}
+			sink := duckv1.Addressable{
+				URL: sinkURI,
 			}
 
 			a := &Adapter{
@@ -348,14 +352,13 @@ func TestPostMessage_ServeHTTP_binary_mode(t *testing.T) {
 					ConsumerGroup: "group",
 					Name:          "test",
 				},
-				httpMessageSender: s,
-				logger:            zap.NewNop().Sugar(),
-				reporter:          statsReporter,
-				keyTypeMapper:     getKeyTypeMapper(tc.keyTypeMapper),
+				sink:          sink,
+				logger:        zap.NewNop().Sugar(),
+				reporter:      statsReporter,
+				keyTypeMapper: getKeyTypeMapper(tc.keyTypeMapper),
 			}
 
 			_, err = a.Handle(context.TODO(), tc.message)
-
 			if tc.error && err == nil {
 				t.Errorf("expected error, but got %v", err)
 			}
@@ -434,7 +437,7 @@ func TestAdapter_Start(t *testing.T) { // just increase code coverage
 	// Increasing coverage
 	_ = os.Setenv("KAFKA_BOOTSTRAP_SERVERS", "my-cluster-kafka-bootstrap.my-kafka-namespace:9092")
 
-	a := NewAdapter(ctx, NewEnvConfig(), nil, nil)
+	a := NewAdapter(ctx, NewEnvConfig(), duckv1.Addressable{}, nil)
 	err := a.Start(ctx)
 	if err == nil {
 		t.Errorf("expected error, but got nil")
diff --git a/pkg/source/mtadapter/adapter_test.go b/pkg/source/mtadapter/adapter_test.go
index 6bca3245c..52392f7c6 100644
--- a/pkg/source/mtadapter/adapter_test.go
+++ b/pkg/source/mtadapter/adapter_test.go
@@ -26,12 +26,12 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 
 	"knative.dev/eventing/pkg/metrics/source"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	pkgtesting "knative.dev/pkg/reconciler/testing"
 
 	"knative.dev/eventing/pkg/adapter/v2"
 	adaptertest "knative.dev/eventing/pkg/adapter/v2/test"
-	"knative.dev/eventing/pkg/kncloudevents"
 
 	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
@@ -376,7 +376,7 @@ type sampleAdapter struct {
 	running bool
 }
 
-func newSampleAdapter(ctx context.Context, env adapter.EnvConfigAccessor, adapter *kncloudevents.HTTPMessageSender, reporter source.StatsReporter) adapter.MessageAdapter {
+func newSampleAdapter(ctx context.Context, env adapter.EnvConfigAccessor, sink duckv1.Addressable, reporter source.StatsReporter) adapter.MessageAdapter {
 	return &sampleAdapter{}
 }
 
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/103-pod.yaml b/vendor/knative.dev/reconciler-test/pkg/eventshub/103-pod.yaml
index 568d19f15..8b591df1c 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/103-pod.yaml
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/103-pod.yaml
@@ -26,7 +26,7 @@ metadata:
       {{ end }}
   {{ end }}
 spec:
-  serviceAccountName: "{{ .namespace }}"
+  serviceAccountName: "{{ .name }}"
   restartPolicy: "OnFailure"
   {{ if .podSecurityContext }}
   securityContext:
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/104-forwarder.yaml b/vendor/knative.dev/reconciler-test/pkg/eventshub/104-forwarder.yaml
index 3534ff20c..a121662e1 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/104-forwarder.yaml
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/104-forwarder.yaml
@@ -32,7 +32,7 @@ spec:
         {{ end }}
     {{ end }}
     spec:
-      serviceAccountName: "{{ .namespace }}"
+      serviceAccountName: "{{ .name }}"
       containers:
         - name: eventshub-forwarder
           image: {{ .image }}
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/forwarder/forwarder.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/forwarder/forwarder.go
index ba940f72d..19ed26e53 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/forwarder/forwarder.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/forwarder/forwarder.go
@@ -29,6 +29,7 @@ import (
 	"go.opencensus.io/trace"
 	"go.uber.org/zap"
 
+	"github.com/cloudevents/sdk-go/v2/binding"
 	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
 	cloudeventshttp "github.com/cloudevents/sdk-go/v2/protocol/http"
 	"github.com/kelseyhightower/envconfig"
@@ -164,7 +165,8 @@ func (o *Forwarder) ServeHTTP(writer http.ResponseWriter, request *http.Request)
 	if err != nil {
 		logging.FromContext(o.ctx).Error("Cannot create the request: ", err)
 	}
-	err = cehttp.WriteRequest(requestCtx, m, req)
+
+	err = cehttp.WriteRequest(requestCtx, binding.ToMessage(event), req)
 	if err != nil {
 		logging.FromContext(o.ctx).Error("Cannot write the event: ", err)
 	}
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/100-sa.yaml b/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/100-sa.yaml
index b94996272..f86b52394 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/100-sa.yaml
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/100-sa.yaml
@@ -15,5 +15,5 @@
 apiVersion: v1
 kind: ServiceAccount
 metadata:
-  name: {{ .namespace }}
+  name: {{ .name }}
   namespace: {{ .namespace }}
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/101-rbac.yaml b/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/101-rbac.yaml
index 5839e77b2..dffe43896 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/101-rbac.yaml
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/101-rbac.yaml
@@ -15,7 +15,7 @@
 apiVersion: rbac.authorization.k8s.io/v1
 kind: Role
 metadata:
-  name: {{ .namespace }}
+  name: {{ .name }}
   namespace: {{ .namespace }}
 rules:
   - apiGroups: [ "" ]
@@ -35,13 +35,13 @@ rules:
 apiVersion: rbac.authorization.k8s.io/v1
 kind: RoleBinding
 metadata:
-  name: {{ .namespace }}
+  name: {{ .name }}
   namespace: {{ .namespace }}
 roleRef:
   apiGroup: rbac.authorization.k8s.io
   kind: Role
-  name: {{ .namespace }}
+  name: {{ .name }}
 subjects:
   - kind: ServiceAccount
-    name: {{ .namespace }}
+    name: {{ .name }}
     namespace: {{ .namespace }}
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/rbac.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/rbac.go
index 099b73a83..de8a2cfbd 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/rbac.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/rbac.go
@@ -31,9 +31,9 @@ var templates embed.FS
 
 // Install creates the necessary ServiceAccount, Role, RoleBinding for the eventshub.
 // The resources are named according to the current namespace defined in the environment.
-func Install() feature.StepFn {
+func Install(cfg map[string]interface{}) feature.StepFn {
 	return func(ctx context.Context, t feature.T) {
-		if _, err := manifest.InstallYamlFS(ctx, templates, map[string]interface{}{}); err != nil && !apierrors.IsAlreadyExists(err) {
+		if _, err := manifest.InstallYamlFS(ctx, templates, cfg); err != nil && !apierrors.IsAlreadyExists(err) {
 			t.Fatal(err)
 		}
 	}
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/resources.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/resources.go
index 3011fc062..3b53a34a5 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/resources.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/resources.go
@@ -86,9 +86,6 @@ func Install(name string, options ...EventsHubOption) feature.StepFn {
 		eventListener := k8s.EventListenerFromContext(ctx)
 		registerEventsHubStore(ctx, eventListener, name, namespace)
 
-		// Install ServiceAccount, Role, RoleBinding
-		eventshubrbac.Install()(ctx, t)
-
 		isReceiver := strings.Contains(envs[EventGeneratorsEnv], "receiver")
 		isEnforceTLS := strings.Contains(envs[EnforceTLS], "true")
 
@@ -114,6 +111,9 @@ func Install(name string, options ...EventsHubOption) feature.StepFn {
 			"withEnforceTLS": isEnforceTLS,
 		}
 
+		// Install ServiceAccount, Role, RoleBinding
+		eventshubrbac.Install(cfg)(ctx, t)
+
 		if ic := environment.GetIstioConfig(ctx); ic.Enabled {
 			manifest.WithIstioPodAnnotations(cfg)
 		}