Repository: knative-sandbox/eventing-kafka-broker
Pull request #2404 (2022-07-12 09:21:04)

Test delivery timeout
Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>


knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2404#" title="Author self-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-broker).

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-broker/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":[]} -->
knative-prow[bot]: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 40783b6c914f4ec3668b873a1f2c0f99964b3b54 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/unit-tests_eventing-kafka-broker_main/1546788601038114816) | true | `/test unit-tests_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 04128af08f03c0cc0412e46402fea7957308b7b6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/unit-tests_eventing-kafka-broker_main/1546792082406903808) | true | `/test unit-tests_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @pierDipi: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 04128af08f03c0cc0412e46402fea7957308b7b6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/unit-tests_eventing-kafka-broker_main/1546792082406903808) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 04128af08f03c0cc0412e46402fea7957308b7b6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1546792089008738304) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @pierDipi: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 04128af08f03c0cc0412e46402fea7957308b7b6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/unit-tests_eventing-kafka-broker_main/1546792082406903808) | true | `/test unit-tests_eventing-kafka-broker_main`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 04128af08f03c0cc0412e46402fea7957308b7b6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1546792089008738304) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 04128af08f03c0cc0412e46402fea7957308b7b6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/integration-tests_eventing-kafka-broker_main/1546792082968940544) | true | `/test integration-tests_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @pierDipi: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 04128af08f03c0cc0412e46402fea7957308b7b6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1546792089008738304) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 04128af08f03c0cc0412e46402fea7957308b7b6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/integration-tests_eventing-kafka-broker_main/1546792082968940544) | true | `/test integration-tests_eventing-kafka-broker_main`
unit-tests_eventing-kafka-broker_main | ed268575712da7359a791ebbdeefabe3a9fa87e2 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/unit-tests_eventing-kafka-broker_main/1554847748279242752) | true | `/test unit-tests_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
pierDipi: /retest-required
knative-prow[bot]: @pierDipi: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 04128af08f03c0cc0412e46402fea7957308b7b6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1546792089008738304) | true | `/test channel-integration-tests-sasl-plain_eventing-kafka-broker_main`
integration-tests_eventing-kafka-broker_main | 04128af08f03c0cc0412e46402fea7957308b7b6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/integration-tests_eventing-kafka-broker_main/1546792082968940544) | true | `/test integration-tests_eventing-kafka-broker_main`
upgrade-tests_eventing-kafka-broker_main | ed268575712da7359a791ebbdeefabe3a9fa87e2 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2404/upgrade-tests_eventing-kafka-broker_main/1554847750141513728) | true | `/test upgrade-tests_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
pierDipi: Experimental test suite passed.
pierDipi: /assign @matzew @aliok 
pierDipi: /retest-required

Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 10b84c7cc..2f14d6006 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -201,6 +201,7 @@ aliases:
   - knative-prow-updater-robot
   - knative-test-reporter-robot
   kperf-approvers:
+  - chzhyang
   - maximilien
   - nader-ziada
   - psschwei
@@ -256,6 +257,11 @@ aliases:
   - kvmware
   - psschwei
   - upodroid
+  security-guard-approvers:
+  - davidhadas
+  - maximilien
+  - psschwei
+  - rhuss
   security-wg-leads:
   - evankanderson
   - julz
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 2235a2b4f..c2286ee0a 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -172,7 +172,8 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		Uid:    string(ks.UID),
 		Topics: []string{ks.Spec.Topic},
 		Ingress: &contract.Ingress{
-			Path: receiver.PathFromObject(ks),
+			Path:        receiver.PathFromObject(ks),
+			ContentMode: coreconfig.ContentModeFromString(*ks.Spec.ContentMode),
 		},
 		BootstrapServers: kafka.BootstrapServersCommaSeparated(ks.Spec.BootstrapServers),
 		Reference: &contract.Reference{
diff --git a/data-plane/profiler/run.sh b/data-plane/profiler/run.sh
index 06726ad9c..070682e7f 100755
--- a/data-plane/profiler/run.sh
+++ b/data-plane/profiler/run.sh
@@ -40,7 +40,7 @@ echo "Profiling event ${EVENT}"
 
 PROJECT_ROOT_DIR=$(dirname $0)/..
 RESOURCES_DIR="$(dirname $0)"/resources
-ASYNC_PROFILER_URL="https://github.com/jvm-profiling-tools/async-profiler/releases/download/v2.6/async-profiler-2.6-linux-x64.tar.gz"
+ASYNC_PROFILER_URL="https://github.com/jvm-profiling-tools/async-profiler/releases/download/v2.8.3/async-profiler-2.8.3-linux-x64.tar.gz"
 KAFKA_URL="https://archive.apache.org/dist/kafka/2.6.0/kafka_2.13-2.6.0.tgz"
 LOG_DIR=${LOG_DIR:-"/tmp/eventing-kafka-broker-logs/profiler"}
 
@@ -101,7 +101,7 @@ export WAIT_STARTUP_SECONDS="8"
 export SERVICE_NAME="kafka-broker-receiver"
 export SERVICE_NAMESPACE="knative-eventing"
 export INGRESS_PORT="8080"
-export METRICS_PORT="9090"
+export METRICS_PORT="9098"
 export INSTANCE_ID="receiver"
 
 # Run receiver.
@@ -116,7 +116,7 @@ receiver_pid=$!
 export SERVICE_NAME="kafka-broker-dispatcher"
 export SERVICE_NAMESPACE="knative-eventing"
 export INGRESS_PORT="8080"
-export METRICS_PORT="9089"
+export METRICS_PORT="9099"
 export INSTANCE_ID="dispatcher"
 
 # Run dispatcher.
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
index 20df63368..872df8190 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
@@ -89,7 +89,7 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
     // That means, we support these modes:
     // - Request coming to "/path" --> path is used for matching
     // - Request coming to "/" --> hostname is used for matching
-    final var p = pathMapper.get(path);
+    final var p = pathMapper.get(removeTrailingSlash(path));
     if (p != null) {
       return p;
     }
@@ -113,6 +113,13 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
     return hostMapper.get(host);
   }
 
+  private String removeTrailingSlash(final String path) {
+    if (path.endsWith("/")) {
+      return path.substring(0, path.length() - 1);
+    }
+    return path;
+  }
+
   @Override
   public Future<Void> onNewIngress(
     DataPlaneContract.Resource resource,
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 1c9b175b3..13b077f81 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1261,6 +1261,7 @@ knative.dev/eventing/test
 knative.dev/eventing/test/conformance/helpers
 knative.dev/eventing/test/conformance/helpers/tracing
 knative.dev/eventing/test/e2e/helpers
+knative.dev/eventing/test/experimental/features/delivery_timeout
 knative.dev/eventing/test/experimental/features/new_trigger_filters
 knative.dev/eventing/test/flags
 knative.dev/eventing/test/lib


Test patch:
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 4c7c06d80..6ef1e2180 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -162,6 +162,60 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal - content mode structured",
+			Objects: []runtime.Object{
+				NewSink(
+					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
+					SinkContentMode(v1alpha1.ModeStructured),
+				),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve": "value_to_preserve",
+				}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, env.DataPlaneConfigFormat, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              SinkUUID,
+							Topics:           []string{SinkTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
+							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
+						},
+					},
+					Generation: 1,
+				}),
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSink(
+						StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
+						SinkContentMode(v1alpha1.ModeStructured),
+						InitSinkConditions,
+						StatusDataPlaneAvailable,
+						StatusConfigParsed,
+						BootstrapServers(bootstrapServersArr),
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
+						SinkAddressable(&env),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+		},
 		{
 			Name: "Reconciled normal - with auth config",
 			Objects: []runtime.Object{
diff --git a/control-plane/pkg/reconciler/testing/objects_sink.go b/control-plane/pkg/reconciler/testing/objects_sink.go
index 7aaea0294..d2cb723b0 100644
--- a/control-plane/pkg/reconciler/testing/objects_sink.go
+++ b/control-plane/pkg/reconciler/testing/objects_sink.go
@@ -166,3 +166,10 @@ func SinkReceiverPodUpdate(namespace string, annotations map[string]string) clie
 		SinkReceiverPod(namespace, annotations),
 	)
 }
+
+func SinkContentMode(cm string) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*eventing.KafkaSink)
+		ks.Spec.ContentMode = &cm
+	}
+}
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
index eecda4cca..9f549995c 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
@@ -422,9 +422,8 @@ public class RecordDispatcherTest {
           .summaries()
           .stream()
           .reduce((a, b) -> b)
-          .get()
-          .max()
-      ).isGreaterThan(0)
+          .isPresent()
+      ).isTrue()
     );
   }
 
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStoreTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStoreTest.java
index 6674aca69..b0321cd94 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStoreTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStoreTest.java
@@ -406,6 +406,7 @@ public class IngressProducerReconcilableStoreTest {
 
           // only use path when the path is registered
           assertThat(store.resolve("http://host1", "/hello1").getKafkaProducer()).isSameAs(producer1);
+          assertThat(store.resolve("http://host1", "/hello1/").getKafkaProducer()).isSameAs(producer1);
           assertThat(store.resolve("http://host2", "/hello1").getKafkaProducer()).isSameAs(producer1);
           assertThat(store.resolve("http://host1", "/hello3").getKafkaProducer()).isSameAs(producer3);
           assertThat(store.resolve("http://host2", "/hello3").getKafkaProducer()).isSameAs(producer3);
diff --git a/test/config/100-config-experimental-features.yaml b/test/config/100-config-experimental-features.yaml
new file mode 120000
index 000000000..bfbbc3e52
--- /dev/null
+++ b/test/config/100-config-experimental-features.yaml
@@ -0,0 +1,1 @@
+../experimental/features_config/features.yaml
\ No newline at end of file
diff --git a/test/e2e_new/broker_conformance_test.go b/test/e2e_new/broker_conformance_test.go
index c99a223b6..47680aff5 100644
--- a/test/e2e_new/broker_conformance_test.go
+++ b/test/e2e_new/broker_conformance_test.go
@@ -33,8 +33,6 @@ import (
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
-
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 )
 
 func TestBrokerConformance(t *testing.T) {
@@ -47,7 +45,7 @@ func TestBrokerConformance(t *testing.T) {
 	)
 
 	// Install and wait for a Ready Broker.
-	env.Prerequisite(ctx, t, brokerfeatures.GoesReady("default", broker.WithBrokerClass(kafka.BrokerClass)))
+	env.Prerequisite(ctx, t, brokerfeatures.GoesReady("default", broker.WithBrokerClass(broker.EnvCfg.BrokerClass)))
 
 	env.TestSet(ctx, t, brokerfeatures.ControlPlaneConformance("default"))
 	env.TestSet(ctx, t, brokerfeatures.DataPlaneConformance("default"))
diff --git a/test/e2e_new/broker_delete_config_map_test.go b/test/e2e_new/broker_delete_config_map_test.go
index 177a28416..152c188c4 100644
--- a/test/e2e_new/broker_delete_config_map_test.go
+++ b/test/e2e_new/broker_delete_config_map_test.go
@@ -34,7 +34,6 @@ import (
 	"knative.dev/reconciler-test/pkg/knative"
 	"knative.dev/reconciler-test/resources/svc"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/test/e2e_new/resources/configmap"
 )
 
@@ -65,7 +64,7 @@ func BrokerDeleteContractConfigMap() *feature.Feature {
 	// Create the broker
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 	))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
 	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
diff --git a/test/e2e_new/broker_not_ready_test.go b/test/e2e_new/broker_not_ready_test.go
index 42948656a..7b13dab4d 100644
--- a/test/e2e_new/broker_not_ready_test.go
+++ b/test/e2e_new/broker_not_ready_test.go
@@ -29,7 +29,6 @@ import (
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	brokerconfigmap "knative.dev/eventing-kafka-broker/test/e2e_new/resources/configmap/broker"
 )
 
@@ -63,7 +62,7 @@ func BrokerNotReadyAfterBeingReady() *feature.Feature {
 
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 		broker.WithConfig(configName),
 	))
 
diff --git a/test/e2e_new/dls_test.go b/test/e2e_new/dls_test.go
index 5afaa0595..23a828981 100644
--- a/test/e2e_new/dls_test.go
+++ b/test/e2e_new/dls_test.go
@@ -35,8 +35,6 @@ import (
 	"knative.dev/reconciler-test/resources/svc"
 
 	. "knative.dev/reconciler-test/pkg/eventshub/assert"
-
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 )
 
 func TestDeadLetterSink(t *testing.T) {
@@ -67,7 +65,7 @@ func SendsEventWithRetries() *feature.Feature {
 
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 	))
 
 	f.Setup("broker is ready", broker.IsReady(brokerName))
@@ -127,7 +125,7 @@ func SendsEventNoRetries() *feature.Feature {
 
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 	))
 
 	f.Setup("broker is ready", broker.IsReady(brokerName))
diff --git a/test/e2e_new/main_test.go b/test/e2e_new/main_test.go
index 75c1f0b01..86df30ac4 100644
--- a/test/e2e_new/main_test.go
+++ b/test/e2e_new/main_test.go
@@ -21,6 +21,7 @@ package e2e_new
 
 import (
 	"flag"
+	"fmt"
 	"log"
 	"os"
 	"testing"
@@ -48,7 +49,9 @@ func init() {
 	environment.InitFlags(flag.CommandLine)
 
 	// Force the broker class to be configured properly
-	broker.EnvCfg.BrokerClass = kafka.BrokerClass
+	if broker.EnvCfg.BrokerClass != kafka.BrokerClass && broker.EnvCfg.BrokerClass != kafka.NamespacedBrokerClass {
+		panic(fmt.Errorf("KafkaBroker class '%s' is unknown. Specify 'BROKER_CLASS' env var", broker.EnvCfg.BrokerClass))
+	}
 }
 
 // TestMain is the first entry point for `go test`.
diff --git a/test/e2e_new/ordered_test.go b/test/e2e_new/ordered_test.go
index a815f001f..1383faee4 100644
--- a/test/e2e_new/ordered_test.go
+++ b/test/e2e_new/ordered_test.go
@@ -40,7 +40,6 @@ import (
 	"knative.dev/reconciler-test/pkg/knative"
 	"knative.dev/reconciler-test/resources/svc"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/test/e2e_new/multiple_partition_config"
 	"knative.dev/eventing-kafka-broker/test/e2e_new/single_partition_config"
 
@@ -77,7 +76,7 @@ func SinglePartitionOrderedDelivery() *feature.Feature {
 	f.Setup("install one partition configuration", single_partition_config.Install)
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 		broker.WithConfig(single_partition_config.ConfigMapName),
 	))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
@@ -133,7 +132,7 @@ func MultiplePartitionOrderedDelivery() *feature.Feature {
 	f.Setup("install multiple partition configuration", multiple_partition_config.Install)
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 		broker.WithConfig(multiple_partition_config.ConfigMapName),
 	))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
diff --git a/test/e2e_new/tracing_test.go b/test/e2e_new/tracing_test.go
index 9c91094a6..dd7242c72 100644
--- a/test/e2e_new/tracing_test.go
+++ b/test/e2e_new/tracing_test.go
@@ -40,7 +40,6 @@ import (
 
 	. "knative.dev/reconciler-test/pkg/eventshub/assert"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/test/pkg/tracing"
 )
 
@@ -75,7 +74,7 @@ func TracingHeadersUsingOrderedDeliveryWithTraceExported() *feature.Feature {
 
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 	))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
 	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
@@ -206,7 +205,7 @@ func TracingHeadersUsingUnorderedDelivery() *feature.Feature {
 
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 	))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
 	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
@@ -252,7 +251,7 @@ func TracingHeadersUsingUnorderedDeliveryWithMultipleTriggers() *feature.Feature
 
 	f.Setup("install broker", broker.Install(
 		brokerName,
-		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
 	))
 	f.Setup("broker is ready", broker.IsReady(brokerName))
 	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
diff --git a/test/experimental/delivery_timeout_test.go b/test/experimental/delivery_timeout_test.go
new file mode 100644
index 000000000..8bb2ebe00
--- /dev/null
+++ b/test/experimental/delivery_timeout_test.go
@@ -0,0 +1,65 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * Copyright 2022 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 experimental
+
+import (
+	"testing"
+
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+
+	"knative.dev/eventing/test/experimental/features/delivery_timeout"
+	"knative.dev/eventing/test/rekt/features/broker"
+	b "knative.dev/eventing/test/rekt/resources/broker"
+	rt "knative.dev/eventing/test/rekt/resources/trigger"
+)
+
+func TestDeliveryTimeout(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, delivery_timeout.ChannelToSink())
+}
+
+func TestBrokerTriggerWithDeliveryTimeout(t *testing.T) {
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	brokerName := feature.MakeRandomK8sName("broker")
+	triggerName := feature.MakeRandomK8sName("trigger")
+
+	env.Test(ctx, t, broker.GoesReady(brokerName, append(b.WithEnvConfig(), b.WithTimeout("PT1S"))...))
+	env.Test(ctx, t, broker.TriggerGoesReady(triggerName, brokerName, rt.WithTimeout("PT10S")))
+}
diff --git a/test/experimental/features_config/features.yaml b/test/experimental/features_config/features.yaml
index dd4b493d4..cf6829924 100644
--- a/test/experimental/features_config/features.yaml
+++ b/test/experimental/features_config/features.yaml
@@ -24,6 +24,6 @@ metadata:
 data:
   kreference-group: "disabled"
   delivery-retryafter: "disabled"
-  delivery-timeout: "disabled"
+  delivery-timeout: "enabled"
   strict-subscriber: "disabled"
   new-trigger-filters: "enabled"
diff --git a/test/experimental/main_test.go b/test/experimental/main_test.go
index f9d653c99..9ab5fdab0 100644
--- a/test/experimental/main_test.go
+++ b/test/experimental/main_test.go
@@ -26,6 +26,7 @@ import (
 
 	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
 	"knative.dev/pkg/injection"
 	_ "knative.dev/pkg/system/testing"
 
@@ -40,6 +41,10 @@ var global environment.GlobalEnvironment
 func init() {
 	// environment.InitFlags registers state and level filter flags.
 	environment.InitFlags(flag.CommandLine)
+
+	// CHANNEL_GROUP_KIND=KafkaChannel.messaging.knative.dev;CHANNEL_VERSION=v1beta1
+	channel_impl.EnvCfg.ChannelGK = "KafkaChannel.messaging.knative.dev"
+	channel_impl.EnvCfg.ChannelV = "v1beta1"
 }
 
 // TestMain is the first entry point for `go test`.
diff --git a/test/reconciler-tests.sh b/test/reconciler-tests.sh
index be8205f0b..c44fefa61 100755
--- a/test/reconciler-tests.sh
+++ b/test/reconciler-tests.sh
@@ -27,6 +27,8 @@ if [ "${EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO:-""}" != "" ]; then
   success
 fi
 
+export BROKER_CLASS="Kafka"
+
 go_test_e2e -timeout=1h ./test/e2e_new/... || fail_test "E2E (new) suite failed"
 
 go_test_e2e -tags=e2e,cloudevents -timeout=1h ./test/e2e_new_channel/... || fail_test "E2E (new - KafkaChannel) suite failed"
diff --git a/vendor/knative.dev/eventing/test/experimental/features/delivery_timeout/channel.go b/vendor/knative.dev/eventing/test/experimental/features/delivery_timeout/channel.go
new file mode 100644
index 000000000..ed8903ff0
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/experimental/features/delivery_timeout/channel.go
@@ -0,0 +1,119 @@
+/*
+Copyright 2021 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 delivery_timeout
+
+import (
+	"context"
+	"time"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/rickb777/date/period"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	eventingclient "knative.dev/eventing/pkg/client/injection/client"
+	"knative.dev/eventing/test/rekt/resources/channel"
+	"knative.dev/eventing/test/rekt/resources/subscription"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/eventshub/assert"
+	"knative.dev/reconciler-test/pkg/feature"
+)
+
+// ChannelToSink tests a scenario where the flow is source -> channel -> sink (timeout) -- fallback to -> dead letter sink
+func ChannelToSink() *feature.Feature {
+	f := feature.NewFeature()
+
+	timeout := 6 * time.Second
+	// Clocks are funny, let's add 1 second to take in account eventual clock skews
+	timeoutPeriod, _ := period.NewOf(timeout - time.Second)
+	timeoutString := timeoutPeriod.String()
+
+	channelAPIVersion, imcKind := channel.GVK().ToAPIVersionAndKind()
+
+	channelName := feature.MakeRandomK8sName("channel")
+	subName := feature.MakeRandomK8sName("sub-sink")
+	sinkName := feature.MakeRandomK8sName("sink")
+	deadLetterSinkName := feature.MakeRandomK8sName("dead-letter")
+	sourceName := feature.MakeRandomK8sName("source")
+
+	ev := cetest.FullEvent()
+
+	f.Setup("install sink", eventshub.Install(
+		sinkName,
+		eventshub.StartReceiver,
+		eventshub.ResponseWaitTime(timeout),
+	))
+	f.Setup("install dead letter sink", eventshub.Install(
+		deadLetterSinkName,
+		eventshub.StartReceiver,
+	))
+
+	f.Setup("Install channel", channel.Install(channelName))
+	f.Setup("Channel is ready", channel.IsReady(channelName))
+
+	f.Setup("Install channel -> sink subscription", func(ctx context.Context, t feature.T) {
+		namespace := environment.FromContext(ctx).Namespace()
+		_, err := eventingclient.Get(ctx).MessagingV1().Subscriptions(namespace).Create(ctx,
+			&messagingv1.Subscription{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      subName,
+					Namespace: namespace,
+				},
+				Spec: messagingv1.SubscriptionSpec{
+					Channel: duckv1.KReference{
+						APIVersion: channelAPIVersion,
+						Kind:       imcKind,
+						Name:       channelName,
+					},
+					Subscriber: &duckv1.Destination{
+						Ref: &duckv1.KReference{
+							APIVersion: "v1",
+							Kind:       "Service",
+							Name:       sinkName,
+						},
+					},
+					Delivery: &eventingduckv1.DeliverySpec{
+						DeadLetterSink: &duckv1.Destination{
+							Ref: &duckv1.KReference{
+								APIVersion: "v1",
+								Kind:       "Service",
+								Name:       deadLetterSinkName,
+							},
+						},
+						Timeout: &timeoutString,
+					},
+				},
+			}, metav1.CreateOptions{})
+		require.NoError(t, err)
+	})
+
+	f.Setup("subscription channel -> Sink is ready", subscription.IsReady(subName))
+
+	f.Setup("install source", eventshub.Install(
+		sourceName,
+		eventshub.StartSenderToResource(channel.GVR(), channelName),
+		eventshub.InputEvent(ev),
+	))
+
+	f.Assert("receive event on sink", assert.OnStore(sinkName).MatchEvent(cetest.HasId(ev.ID())).Exact(1))
+	f.Assert("receive event on dead letter sink", assert.OnStore(deadLetterSinkName).MatchEvent(cetest.HasId(ev.ID())).Exact(1))
+
+	return f
+}