[0.17] Port recordevents

go.sum

@@ -373,6 +373,7 @@ github.com/evanphx/json-patch v4.5.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLi
 github.com/exponent-io/jsonpath v0.0.0-20151013193312-d6023ce2651d/go.mod h1:ZZMPRZwes7CROmyNKgQzC3XPs6L/G2EJLHddWejkmf4=
 github.com/fatih/camelcase v1.0.0/go.mod h1:yN2Sb0lFhZJUdVvtELVWefmrXpuZESvPmqwoZc+/fpc=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
+github.com/fatih/color v1.9.0 h1:8xPHl4/q1VyqGIPif1F+1V3Y3lSmrq01EabUW3CoW5s=
 github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
 github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568/go.mod h1:xEzjJPgXI435gkrCt3MPfRiAkVrwSbHsst4LCFVfpJc=
 github.com/fortytw2/leaktest v1.2.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
@@ -773,6 +774,7 @@ github.com/jmoiron/sqlx v1.2.1-0.20190826204134-d7d95172beb5/go.mod h1:1FEQNm3xl
 github.com/joefitzgerald/rainbow-reporter v0.1.0/go.mod h1:481CNgqmVHQZzdIbN52CupLJyoVwB10FQ/IQlF1pdL8=
 github.com/joho/godotenv v1.3.0 h1:Zjp+RcGpHhGlrMbJzXTrZZPrWj+1vfm90La1wgB6Bhc=
 github.com/joho/godotenv v1.3.0/go.mod h1:7hK45KPybAkOC6peb+G5yklZfMxEjkZhHbwpqxOKXbg=
+github.com/jonboulle/clockwork v0.1.0 h1:VKV+ZcuP6l3yW9doeqz6ziZGgcynBVQO+obU0+0hcPo=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
 github.com/jpillora/backoff v0.0.0-20180909062703-3050d21c67d7/go.mod h1:2iMrUgbbvHEiQClaW2NsSzMyGHqN+rDFqY705q49KG0=
 github.com/json-iterator/go v0.0.0-20180612202835-f2b4162afba3/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
@@ -854,6 +856,7 @@ github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaO
 github.com/mattn/go-colorable v0.1.1/go.mod h1:FuOcm+DKB9mbwrcAfNl7/TZVBZ6rcnceauSikq3lYCQ=
 github.com/mattn/go-colorable v0.1.2/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
 github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
+github.com/mattn/go-colorable v0.1.6 h1:6Su7aK7lXmJ/U79bYtBjLNaha4Fs1Rg9plHpcH+vvnE=
 github.com/mattn/go-colorable v0.1.6/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-ieproxy v0.0.0-20190610004146-91bb50d98149/go.mod h1:31jz6HNzdxOmlERGGEc4v/dMssOfmp2p5bT/okiKFFc=
 github.com/mattn/go-ieproxy v0.0.0-20190702010315-6dee0af9227d/go.mod h1:31jz6HNzdxOmlERGGEc4v/dMssOfmp2p5bT/okiKFFc=
@@ -863,6 +866,7 @@ github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNx
 github.com/mattn/go-isatty v0.0.5/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/mattn/go-isatty v0.0.11/go.mod h1:PhnuNfih5lzO57/f3n+odYbM4JtupLOxQOAqxQCu2WE=
+github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY=
 github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
 github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
 github.com/mattn/go-runewidth v0.0.8/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
@@ -1107,6 +1111,7 @@ github.com/spf13/cobra v0.0.2-0.20171109065643-2da4a54c5cee/go.mod h1:1l0Ry5zgKv
 github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ=
 github.com/spf13/cobra v0.0.5/go.mod h1:3K3wKZymM7VvHMDS9+Akkh4K60UwM26emMESw8tLCHU=
 github.com/spf13/cobra v0.0.6/go.mod h1:/6GTrnGXV9HjY+aR4k0oJ5tcvakLuG6EuKReYlHNrgE=
+github.com/spf13/cobra v1.0.0 h1:6m/oheQuQ13N9ks4hubMG6BnvwOeaJrqSPLahSnczz8=
 github.com/spf13/cobra v1.0.0/go.mod h1:/6GTrnGXV9HjY+aR4k0oJ5tcvakLuG6EuKReYlHNrgE=
 github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo=
 github.com/spf13/jwalterweatherman v1.1.0/go.mod h1:aNWZUN0dPAAO/Ljvb5BEdw96iTZ0EXowPYD95IqWIGo=
@@ -1760,6 +1765,7 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20190709130402-674ba3eaed22/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20191026110619-0b21df46bc1d h1:LCPbGQ34PMrwad11aMZ+dbz5SAsq/0ySjRwQ8I9Qwd8=
 gopkg.in/yaml.v3 v3.0.0-20191026110619-0b21df46bc1d/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=
 gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
 helm.sh/helm/v3 v3.1.1/go.mod h1:WYsFJuMASa/4XUqLyv54s0U/f3mlAaRErGmyy4z921g=
 honnef.co/go/tools v0.0.0-20180728063816-88497007e858/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

test/conformance/helpers/broker_control_plane_test_helper.go

@@ -24,6 +24,7 @@ import (
 	"knative.dev/pkg/reconciler"
 
 	eventingv1beta1 "knative.dev/eventing/pkg/apis/eventing/v1beta1"
+	"knative.dev/eventing/test/lib/recordevents"
 
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/duck"
@@ -81,8 +82,7 @@ func triggerV1Beta1BeforeBrokerHelper(triggerName string, client *testlib.Client
 	const etLogger = "logger"
 	const loggerPodName = "logger-pod"
 
-	logPod := resources.EventRecordPod(loggerPodName)
-	client.CreatePodOrFail(logPod, testlib.WithService(loggerPodName))
+	_ = recordevents.DeployEventRecordOrFail(client, loggerPodName)
 	client.WaitForAllTestResourcesReadyOrFail() // Can't do this for the trigger because it's not 'ready' yet
 	client.CreateTriggerOrFailV1Beta1(triggerName,
 		resources.WithAttributesTriggerFilterV1Beta1(eventingv1beta1.TriggerAnyFilter, etLogger, map[string]interface{}{}),

test/conformance/helpers/broker_data_plane_test_helper.go

@@ -316,13 +316,15 @@ func BrokerV1Beta1ConsumerDataPlaneTestHelper(
 		source := "origin-for-reply"
 		event.SetSource(source)
 		msg := []byte(`{"msg":"Transformed!"}`)
-		transformPod := resources.EventTransformationPod(
+		recordevents.DeployEventRecordOrFail(
+			client,
 			"transformer-pod",
-			"reply-check-type",
-			"reply-check-source",
-			msg,
+			recordevents.ReplyWithTransformedEvent(
+				"reply-check-type",
+				"reply-check-source",
+				string(msg),
+			),
 		)
-		client.CreatePodOrFail(transformPod, testlib.WithService("transformer-pod"))
 		client.WaitForServiceEndpointsOrFail("transformer-pod", 1)
 		transformTrigger := client.CreateTriggerOrFailV1Beta1(
 			"transform-trigger",

test/conformance/helpers/broker_tracing_test_helper.go

@@ -29,6 +29,7 @@ import (
 	"knative.dev/eventing/pkg/utils"
 	tracinghelper "knative.dev/eventing/test/conformance/helpers/tracing"
 	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
 	"knative.dev/eventing/test/lib/sender"
 )
@@ -78,8 +79,7 @@ func setupBrokerTracing(brokerClass string) SetupTracingTestInfrastructureFunc {
 		)
 
 		// Create a logger (EventRecord) Pod and a K8s Service that points to it.
-		logPod := resources.EventRecordPod(loggerPodName)
-		client.CreatePodOrFail(logPod, testlib.WithService(loggerPodName))
+		_ = recordevents.DeployEventRecordOrFail(client, loggerPodName)
 
 		// Create a Trigger that receives events (type=bar) and sends them to the logger Pod.
 		loggerTrigger := client.CreateTriggerOrFailV1Beta1(
@@ -89,15 +89,17 @@ func setupBrokerTracing(brokerClass string) SetupTracingTestInfrastructureFunc {
 			resources.WithSubscriberServiceRefForTriggerV1Beta1(loggerPodName),
 		)
 
-		// Create a transformer (EventTransfrmer) Pod that replies with the same event as the input,
+		// Create a transformer Pod (recordevents with transform reply) that replies with the same event as the input,
 		// except the reply's event's type is changed to bar.
-		eventTransformerPod := resources.EventTransformationPod(
+		eventTransformerPod := recordevents.DeployEventRecordOrFail(
+			client,
 			"transformer",
-			etLogger,
-			senderName,
-			[]byte(eventBody),
+			recordevents.ReplyWithTransformedEvent(
+				etLogger,
+				senderName,
+				eventBody,
+			),
 		)
-		client.CreatePodOrFail(eventTransformerPod, testlib.WithService(eventTransformerPod.Name))
 
 		// Create a Trigger that receives events (type=foo) and sends them to the transformer Pod.
 		transformerTrigger := client.CreateTriggerOrFailV1Beta1(

test/conformance/helpers/channel_status_subscriber_test_helper.go

@@ -22,6 +22,7 @@ import (
 	duckv1beta1 "knative.dev/eventing/pkg/apis/duck/v1beta1"
 	eventingv1beta1 "knative.dev/eventing/pkg/apis/messaging/v1beta1"
 	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
 
 	corev1 "k8s.io/api/core/v1"
@@ -56,8 +57,7 @@ func channelHasRequiredSubscriberStatus(st *testing.T, client *testlib.Client, c
 	client.CreateChannelOrFail(channelName, &channel)
 	client.WaitForResourceReadyOrFail(channelName, &channel)
 
-	pod := resources.EventRecordPod(subscriberServiceName + "-pod")
-	client.CreatePodOrFail(pod, testlib.WithService(subscriberServiceName))
+	_ = recordevents.DeployEventRecordOrFail(client, subscriberServiceName+"-pod")
 
 	subscription := client.CreateSubscriptionOrFail(
 		subscriberServiceName,

test/conformance/helpers/channel_tracing_test_helper.go

@@ -28,6 +28,7 @@ import (
 
 	tracinghelper "knative.dev/eventing/test/conformance/helpers/tracing"
 	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
 	"knative.dev/eventing/test/lib/sender"
 )
@@ -66,17 +67,18 @@ func setupChannelTracingWithReply(
 	client.CreateChannelOrFail(replyChannelName, channel)
 
 	// Create the 'sink', a LogEvents Pod and a K8s Service that points to it.
-	recordEventsPod := resources.EventRecordPod(recordEventsPodName)
-	client.CreatePodOrFail(recordEventsPod, testlib.WithService(recordEventsPodName))
+	recordEventsPod := recordevents.DeployEventRecordOrFail(client, recordEventsPodName)
 
 	// Create the subscriber, a Pod that mutates the event.
-	transformerPod := resources.EventTransformationPod(
+	transformerPod := recordevents.DeployEventRecordOrFail(
+		client,
 		"transformer",
-		"mutated",
-		eventSource,
-		nil,
+		recordevents.ReplyWithTransformedEvent(
+			"mutated",
+			eventSource,
+			"",
+		),
 	)
-	client.CreatePodOrFail(transformerPod, testlib.WithService(transformerPod.Name))
 
 	// Create the Subscription linking the Channel to the mutator.
 	client.CreateSubscriptionOrFail(

test/conformance/helpers/tracing_test_helper.go

@@ -65,7 +65,7 @@ func tracingTest(
 	expectedTestSpan, eventMatcher := setupInfrastructure(t, &channel, client, recordEventsPodName, true)
 
 	// Start the event info store and assert the event was received correctly
-	targetTracker, err := recordevents.NewEventInfoStore(client, recordEventsPodName)
+	targetTracker, err := recordevents.NewEventInfoStore(client, recordEventsPodName, client.Namespace)
 	if err != nil {
 		t.Fatalf("Pod tracker failed: %v", err)
 	}
@@ -105,6 +105,6 @@ func getTraceIDHeader(t *testing.T, evInfos []recordevents.EventInfo) string {
 			}
 		}
 	}
-	t.Fatalf("FAIL: No traceid in %d messages: (%s)", len(evInfos), evInfos)
+	t.Fatalf("FAIL: No traceid in %d messages: (%v)", len(evInfos), evInfos)
 	return ""
 }

test/e2e/helpers/broker_channel_flow_helper.go

@@ -104,13 +104,15 @@ func BrokerChannelFlowWithTransformation(t *testing.T,
 		}
 
 		// create the transformation service for trigger1
-		transformationPod := resources.EventTransformationPod(
+		recordevents.DeployEventRecordOrFail(
+			client,
 			transformationPodName,
-			transformedEventType,
-			transformedEventSource,
-			[]byte(transformedBody),
+			recordevents.ReplyWithTransformedEvent(
+				transformedEventType,
+				transformedEventSource,
+				transformedBody,
+			),
 		)
-		client.CreatePodOrFail(transformationPod, testlib.WithService(transformationPodName))
 
 		// create trigger1 to receive the original event, and do event transformation
 		if triggerVersion == "v1" {

test/e2e/helpers/broker_event_transformation_test_helper.go

@@ -70,13 +70,15 @@ func EventTransformationForTriggerTestHelper(t *testing.T,
 	client.WaitForResourceReadyOrFail(brokerName, testlib.BrokerTypeMeta)
 
 	// create the transformation service
-	transformationPod := resources.EventTransformationPod(
+	recordevents.DeployEventRecordOrFail(
+		client,
 		transformationPodName,
-		transformedEventType,
-		transformedEventSource,
-		[]byte(transformedBody),
+		recordevents.ReplyWithTransformedEvent(
+			transformedEventType,
+			transformedEventSource,
+			transformedBody,
+		),
 	)
-	client.CreatePodOrFail(transformationPod, testlib.WithService(transformationPodName))
 
 	// create trigger1 for event transformation
 	if triggerVersion == "v1" {

test/e2e/helpers/channel_event_tranformation_test_helper.go

@@ -62,13 +62,15 @@ func EventTransformationForSubscriptionTestHelper(t *testing.T,
 		if err := eventAfterTransformation.SetData(cloudevents.ApplicationJSON, []byte(transformedEventBody)); err != nil {
 			t.Fatalf("Cannot set the payload of the event: %s", err.Error())
 		}
-		transformationPod := resources.EventTransformationPod(
+		recordevents.DeployEventRecordOrFail(
+			client,
 			transformationPodName,
-			eventAfterTransformation.Type(),
-			eventAfterTransformation.Source(),
-			eventAfterTransformation.Data(),
+			recordevents.ReplyWithTransformedEvent(
+				eventAfterTransformation.Type(),
+				eventAfterTransformation.Source(),
+				string(eventAfterTransformation.Data()),
+			),
 		)
-		client.CreatePodOrFail(transformationPod, testlib.WithService(transformationPodName))
 
 		// create event logger pod and service as the subscriber
 		eventTracker, _ := recordevents.StartEventRecordOrFail(client, recordEventsPodName)

test/e2e/helpers/parallel_test_helper.go

@@ -71,13 +71,19 @@ func ParallelTestHelper(t *testing.T,
 			for branchNumber, cse := range tc.branchesConfig {
 				// construct filter services
 				filterPodName := fmt.Sprintf("parallel-%s-branch-%d-filter", tc.name, branchNumber)
-				filterPod := resources.EventFilteringPod(filterPodName, cse.filter)
-				client.CreatePodOrFail(filterPod, testlib.WithService(filterPodName))
+				if cse.filter {
+					recordevents.DeployEventRecordOrFail(client, filterPodName)
+				} else {
+					recordevents.DeployEventRecordOrFail(client, filterPodName, recordevents.EchoEvent)
+				}
 
 				// construct branch subscriber
 				subPodName := fmt.Sprintf("parallel-%s-branch-%d-sub", tc.name, branchNumber)
-				subPod := resources.SequenceStepperPod(subPodName, subPodName)
-				client.CreatePodOrFail(subPod, testlib.WithService(subPodName))
+				recordevents.DeployEventRecordOrFail(
+					client,
+					subPodName,
+					recordevents.ReplyWithAppendedData(subPodName),
+				)
 
 				parallelBranches[branchNumber] = v1beta1.ParallelBranch{
 					Filter: &duckv1.Destination{
@@ -177,13 +183,19 @@ func ParallelV1TestHelper(t *testing.T,
 			for branchNumber, cse := range tc.branchesConfig {
 				// construct filter services
 				filterPodName := fmt.Sprintf("parallel-%s-branch-%d-filter", tc.name, branchNumber)
-				filterPod := resources.EventFilteringPod(filterPodName, cse.filter)
-				client.CreatePodOrFail(filterPod, testlib.WithService(filterPodName))
+				if cse.filter {
+					recordevents.DeployEventRecordOrFail(client, filterPodName)
+				} else {
+					recordevents.DeployEventRecordOrFail(client, filterPodName, recordevents.EchoEvent)
+				}
 
 				// construct branch subscriber
 				subPodName := fmt.Sprintf("parallel-%s-branch-%d-sub", tc.name, branchNumber)
-				subPod := resources.SequenceStepperPod(subPodName, subPodName)
-				client.CreatePodOrFail(subPod, testlib.WithService(subPodName))
+				recordevents.DeployEventRecordOrFail(
+					client,
+					subPodName,
+					recordevents.ReplyWithAppendedData(subPodName),
+				)
 
 				parallelBranches[branchNumber] = flowsv1.ParallelBranch{
 					Filter: &duckv1.Destination{

test/e2e/helpers/sequence_test_helper.go

@@ -73,9 +73,11 @@ func SequenceTestHelper(t *testing.T,
 			// create a stepper Pod with Service
 			podName := config.podName
 			msgAppender := config.msgAppender
-			stepperPod := resources.SequenceStepperPod(podName, msgAppender)
+			recordevents.DeployEventRecordOrFail(
+				client, podName,
+				recordevents.ReplyWithAppendedData(msgAppender),
+			)
 
-			client.CreatePodOrFail(stepperPod, testlib.WithService(podName))
 			// create a new step
 			step := v1beta1.SequenceStep{
 				Destination: duckv1.Destination{
@@ -128,8 +130,7 @@ func SequenceTestHelper(t *testing.T,
 		event.SetSource(eventSource)
 		event.SetType(testlib.DefaultEventType)
 		msg := fmt.Sprintf("TestSequence %s", uuid.New().String())
-		body := fmt.Sprintf(`{"msg":"%s"}`, msg)
-		if err := event.SetData(cloudevents.ApplicationJSON, []byte(body)); err != nil {
+		if err := event.SetData(cloudevents.TextPlain, msg); err != nil {
 			st.Fatalf("Cannot set the payload of the event: %s", err.Error())
 		}
 		client.SendEventToAddressable(
@@ -145,7 +146,8 @@ func SequenceTestHelper(t *testing.T,
 		}
 		eventTracker.AssertAtLeast(1, recordevents.MatchEvent(
 			cetest.HasSource(eventSource),
-			cetest.DataContains(expectedMsg),
+			cetest.HasDataContentType(cloudevents.TextPlain),
+			cetest.HasData([]byte(expectedMsg)),
 		))
 	})
 }
@@ -186,9 +188,11 @@ func SequenceV1TestHelper(t *testing.T,
 			// create a stepper Pod with Service
 			podName := config.podName
 			msgAppender := config.msgAppender
-			stepperPod := resources.SequenceStepperPod(podName, msgAppender)
+			recordevents.DeployEventRecordOrFail(
+				client, podName,
+				recordevents.ReplyWithAppendedData(msgAppender),
+			)
 
-			client.CreatePodOrFail(stepperPod, testlib.WithService(podName))
 			// create a new step
 			step := flowsv1.SequenceStep{
 				Destination: duckv1.Destination{
@@ -241,8 +245,7 @@ func SequenceV1TestHelper(t *testing.T,
 		event.SetSource(eventSource)
 		event.SetType(testlib.DefaultEventType)
 		msg := fmt.Sprintf("TestSequence %s", uuid.New().String())
-		body := fmt.Sprintf(`{"msg":"%s"}`, msg)
-		if err := event.SetData(cloudevents.ApplicationJSON, []byte(body)); err != nil {
+		if err := event.SetData(cloudevents.TextPlain, msg); err != nil {
 			st.Fatalf("Cannot set the payload of the event: %s", err.Error())
 		}
 		client.SendEventToAddressable(
@@ -258,7 +261,8 @@ func SequenceV1TestHelper(t *testing.T,
 		}
 		eventTracker.AssertAtLeast(1, recordevents.MatchEvent(
 			cetest.HasSource(eventSource),
-			cetest.DataContains(expectedMsg),
+			cetest.HasDataContentType(cloudevents.TextPlain),
+			cetest.HasData([]byte(expectedMsg)),
 		))
 	})
 }

test/e2e/helpers/trigger_no_broker_test_helper.go

@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 
 	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
 )
 
@@ -40,8 +41,8 @@ func TestTriggerNoBroker(t *testing.T, channel string, brokerCreator BrokerCreat
 	brokerName := strings.ToLower(channel)
 
 	subscriberName := "dumper-empty"
-	eventRecordPod := resources.EventRecordPod(subscriberName)
-	client.CreatePodOrFail(eventRecordPod, testlib.WithService(subscriberName))
+	recordevents.DeployEventRecordOrFail(client, subscriberName)
+
 	client.CreateTriggerOrFailV1Beta1("testtrigger",
 		resources.WithSubscriberServiceRefForTriggerV1Beta1(subscriberName),
 		resources.WithBrokerV1Beta1(brokerName),