Repository: knative-sandbox/eventing-kafka-broker
Pull request #2661 (2022-09-19 16:40:50)

[release-1.7] Metrics for KafkaChannel subscriptions
This is an automated cherry-pick of #2651

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2661?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2661](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2661?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (4aa915e) into [release-1.7](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/8d0bed35f2006f3cf9f1b2d30df87bd820447cc6?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (8d0bed3) will **increase** coverage by `0.08%`.
> The diff coverage is `72.72%`.

```diff
@@               Coverage Diff               @@
##           release-1.7    #2661      +/-   ##
===============================================
+ Coverage        56.12%   56.20%   +0.08%     
===============================================
  Files               74       74              
  Lines             7694     7716      +22     
===============================================
+ Hits              4318     4337      +19     
- Misses            3077     3079       +2     
- Partials           299      300       +1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2661?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2661/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | `70.21% <71.42%> (+0.05%)` | :arrow_up: |
| [control-plane/pkg/reconciler/channel/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2661/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NvbnRyb2xsZXIuZ28=) | `89.15% <100.00%> (+0.13%)` | :arrow_up: |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2661/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `71.95% <0.00%> (+1.10%)` | :arrow_up: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

knative-prow[bot]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_release-1.7 | 4aa915ec03fe27c7e73aee9d39b889b1438e3e30 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2661/reconciler-tests_eventing-kafka-broker_release-1.7/1571902154694725632) | true | `/test reconciler-tests`

[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]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_release-1.7 | 4aa915ec03fe27c7e73aee9d39b889b1438e3e30 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2661/reconciler-tests_eventing-kafka-broker_release-1.7/1571902154694725632) | true | `/test reconciler-tests`
reconciler-tests-namespaced-broker_eventing-kafka-broker_release-1.7 | 4aa915ec03fe27c7e73aee9d39b889b1438e3e30 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2661/reconciler-tests-namespaced-broker_eventing-kafka-broker_release-1.7/1571902154753445888) | true | `/test reconciler-tests-namespaced-broker`

[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]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_release-1.7 | 4aa915ec03fe27c7e73aee9d39b889b1438e3e30 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2661/reconciler-tests_eventing-kafka-broker_release-1.7/1571902154694725632) | true | `/test reconciler-tests`
reconciler-tests-namespaced-broker_eventing-kafka-broker_release-1.7 | 4aa915ec03fe27c7e73aee9d39b889b1438e3e30 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2661/reconciler-tests-namespaced-broker_eventing-kafka-broker_release-1.7/1571902154753445888) | true | `/test reconciler-tests-namespaced-broker`
integration-tests-namespaced-broker_eventing-kafka-broker_release-1.7 | 4aa915ec03fe27c7e73aee9d39b889b1438e3e30 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2661/integration-tests-namespaced-broker_eventing-kafka-broker_release-1.7/1571902154547924992) | true | `/test integration-tests-namespaced-broker`

[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]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_release-1.7 | 4aa915ec03fe27c7e73aee9d39b889b1438e3e30 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2661/reconciler-tests_eventing-kafka-broker_release-1.7/1571902154694725632) | true | `/test reconciler-tests`
reconciler-tests-namespaced-broker_eventing-kafka-broker_release-1.7 | 4aa915ec03fe27c7e73aee9d39b889b1438e3e30 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2661/reconciler-tests-namespaced-broker_eventing-kafka-broker_release-1.7/1571902154753445888) | true | `/test reconciler-tests-namespaced-broker`
integration-tests-namespaced-broker_eventing-kafka-broker_release-1.7 | 4aa915ec03fe27c7e73aee9d39b889b1438e3e30 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2661/integration-tests-namespaced-broker_eventing-kafka-broker_release-1.7/1571902154547924992) | true | `/test integration-tests-namespaced-broker`
channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.7 | 4aa915ec03fe27c7e73aee9d39b889b1438e3e30 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2661/channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.7/1571902154954772480) | true | `/test channel-integration-tests-sasl-plain`

[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]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2661#" title="Author self-approved">knative-prow-robot</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2661#pullrequestreview-1113627636" title="Approved">matzew</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/release-1.7/OWNERS)~~ [matzew]

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/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 2cab70d39..ad007f5fd 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -26,12 +26,13 @@ import (
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	messaging "knative.dev/eventing/pkg/apis/messaging/v1"
 	"knative.dev/pkg/network"
 
 	"github.com/Shopify/sarama"
-
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/labels"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -39,6 +40,7 @@ import (
 	"k8s.io/client-go/util/retry"
 
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
@@ -84,8 +86,9 @@ type Reconciler struct {
 	// reconciliation loop.
 	InitOffsetsFunc kafka.InitOffsetsFunc
 
-	ConfigMapLister corelisters.ConfigMapLister
-	ServiceLister   corelisters.ServiceLister
+	ConfigMapLister    corelisters.ConfigMapLister
+	ServiceLister      corelisters.ServiceLister
+	SubscriptionLister messaginglisters.SubscriptionLister
 
 	Prober prober.Prober
 
@@ -512,6 +515,10 @@ func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messaging
 	if subscriberURI == "" {
 		return nil, fmt.Errorf("failed to resolve Subscription.Spec.Subscriber: empty subscriber URI")
 	}
+	subscriptionName, err := r.getSubscriptionName(channel, subscriber)
+	if err != nil && !apierrors.IsNotFound(err) {
+		return nil, fmt.Errorf("failed to extract subscription name for subscriber %v: %w", subscriber, err)
+	}
 
 	egress := &contract.Egress{
 		Destination:   subscriberURI,
@@ -521,6 +528,14 @@ func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messaging
 		ReplyStrategy: &contract.Egress_DiscardReply{},
 	}
 
+	if subscriptionName != "" {
+		egress.Reference = &contract.Reference{
+			Uuid:      string(subscriber.UID),
+			Namespace: channel.GetNamespace(),
+			Name:      subscriptionName,
+		}
+	}
+
 	if subscriber.ReplyURI != nil {
 		egress.ReplyStrategy = &contract.Egress_ReplyUrl{
 			ReplyUrl: subscriber.ReplyURI.String(),
@@ -641,6 +656,21 @@ func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messa
 	return expected, nil
 }
 
+func (r *Reconciler) getSubscriptionName(channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (string, error) {
+	subscriptions, err := r.SubscriptionLister.Subscriptions(channel.GetNamespace()).List(labels.Everything())
+	if err != nil {
+		return "", fmt.Errorf("failed to list subscriptions in namespace %s: %w", channel.GetNamespace(), err)
+	}
+
+	for _, s := range subscriptions {
+		if s.UID == subscriber.UID {
+			return s.Name, nil
+		}
+	}
+
+	return "", apierrors.NewNotFound(messaging.SchemeGroupVersion.WithResource("subscriptions").GroupResource(), string(subscriber.UID))
+}
+
 // consumerGroup returns a consumerGroup name for the given channel and subscription
 func consumerGroup(channel *messagingv1beta1.KafkaChannel, sub *v1.SubscriberSpec) string {
 	return fmt.Sprintf("kafka.%s.%s.%s", channel.Namespace, channel.Name, string(sub.UID))
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index d91a7b5fd..8e5bfb234 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -25,6 +25,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
 
+	subscriptioninformer "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
+
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	kafkachannelinformer "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
 	kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
@@ -65,6 +67,7 @@ func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 			DispatcherLabel:             base.ChannelDispatcherLabel,
 			ReceiverLabel:               base.ChannelReceiverLabel,
 		},
+		SubscriptionLister:         subscriptioninformer.Get(ctx).Lister(),
 		NewKafkaClient:             sarama.NewClient,
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		InitOffsetsFunc:            offset.InitOffsets,
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
index e3fa0b59e..15c87fe62 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
@@ -77,6 +77,7 @@ public class RecordDispatcherImpl implements RecordDispatcher {
   private static final String KN_ERROR_DEST_EXT_NAME = "knativeerrordest";
   private static final String KN_ERROR_CODE_EXT_NAME = "knativeerrorcode";
   private static final String KN_ERROR_DATA_EXT_NAME = "knativeerrordata";
+  private static final String EKB_ERROR_PREFIX       = "kne-";
   private static final int KN_ERROR_DATA_MAX_BYTES = 1024;
 
   private final Filter filter;
@@ -292,6 +293,13 @@ public class RecordDispatcherImpl implements RecordDispatcher {
     extensions.put(KN_ERROR_DEST_EXT_NAME, destination);
     extensions.put(KN_ERROR_CODE_EXT_NAME, String.valueOf(response.statusCode()));
 
+    // match for prefixed headers and put them to our extensions map
+    response.headers().forEach((k, v) -> {
+      if (k.regionMatches(true, 0, EKB_ERROR_PREFIX, 0, EKB_ERROR_PREFIX.length())) { // aka startsWithIgnoreCase
+        extensions.put(k.substring(EKB_ERROR_PREFIX.length()).toLowerCase(), v);
+      }
+    });
+
     // we extract the response as byte array as we do not need a string
     // representation of it
     var data = response.bodyAsBuffer();
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
new file mode 100644
index 000000000..eb18a1fac
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
@@ -0,0 +1,40 @@
+/*
+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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	fake "knative.dev/eventing/pkg/client/injection/informers/factory/fake"
+	subscription "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = subscription.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Messaging().V1().Subscriptions()
+	return context.WithValue(ctx, subscription.Key{}, inf), inf.Informer()
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
new file mode 100644
index 000000000..37862d8a5
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
@@ -0,0 +1,116 @@
+/*
+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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package subscription
+
+import (
+	context "context"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	cache "k8s.io/client-go/tools/cache"
+	apismessagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	versioned "knative.dev/eventing/pkg/client/clientset/versioned"
+	v1 "knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1"
+	client "knative.dev/eventing/pkg/client/injection/client"
+	factory "knative.dev/eventing/pkg/client/injection/informers/factory"
+	messagingv1 "knative.dev/eventing/pkg/client/listers/messaging/v1"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformer(withInformer)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := factory.Get(ctx)
+	inf := f.Messaging().V1().Subscriptions()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	inf := &wrapper{client: client.Get(ctx), resourceVersion: injection.GetResourceVersion(ctx)}
+	return context.WithValue(ctx, Key{}, inf)
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1.SubscriptionInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1.SubscriptionInformer from context.")
+	}
+	return untyped.(v1.SubscriptionInformer)
+}
+
+type wrapper struct {
+	client versioned.Interface
+
+	namespace string
+
+	resourceVersion string
+}
+
+var _ v1.SubscriptionInformer = (*wrapper)(nil)
+var _ messagingv1.SubscriptionLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apismessagingv1.Subscription{}, 0, nil)
+}
+
+func (w *wrapper) Lister() messagingv1.SubscriptionLister {
+	return w
+}
+
+func (w *wrapper) Subscriptions(namespace string) messagingv1.SubscriptionNamespaceLister {
+	return &wrapper{client: w.client, namespace: namespace, resourceVersion: w.resourceVersion}
+}
+
+// SetResourceVersion allows consumers to adjust the minimum resourceVersion
+// used by the underlying client.  It is not accessible via the standard
+// lister interface, but can be accessed through a user-defined interface and
+// an implementation check e.g. rvs, ok := foo.(ResourceVersionSetter)
+func (w *wrapper) SetResourceVersion(resourceVersion string) {
+	w.resourceVersion = resourceVersion
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apismessagingv1.Subscription, err error) {
+	lo, err := w.client.MessagingV1().Subscriptions(w.namespace).List(context.TODO(), metav1.ListOptions{
+		LabelSelector:   selector.String(),
+		ResourceVersion: w.resourceVersion,
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apismessagingv1.Subscription, error) {
+	return w.client.MessagingV1().Subscriptions(w.namespace).Get(context.TODO(), name, metav1.GetOptions{
+		ResourceVersion: w.resourceVersion,
+	})
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 018bc5cbc..8a2185101 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1340,6 +1340,8 @@ knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger
 knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake
 knative.dev/eventing/pkg/client/injection/informers/factory
 knative.dev/eventing/pkg/client/injection/informers/factory/fake
+knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription
+knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger
 knative.dev/eventing/pkg/client/listers/eventing/v1


Test patch:
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index 41adb3899..e61d09df0 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/Shopify/sarama"
 	"k8s.io/utils/pointer"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingrekttesting "knative.dev/eventing/pkg/reconciler/testing/v1"
 	"knative.dev/pkg/network"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
@@ -439,6 +440,9 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -492,6 +496,79 @@ func TestReconcileKind(t *testing.T) {
 				),
 			},
 		},
+		{
+			Name: "Reconciled normal - with single fresh subscriber - no auth - no subscription",
+			Objects: []runtime.Object{
+				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber))),
+				NewService(),
+				NewPerChannelService(DefaultEnv),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
+							Ingress: &contract.Ingress{
+								Host: receiver.Host(ChannelNamespace, ChannelName),
+							},
+							Egresses: []*contract.Egress{{
+								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
+								Destination:   "http://" + Subscription1URI,
+								Uid:           Subscription1UUID,
+								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+							}},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						ChannelAddressable(&env),
+						WithSubscribers(Subscriber1()),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
 		{
 			Name: "Reconciled normal - with single fresh subscriber - no auth",
 			Objects: []runtime.Object{
@@ -510,6 +587,9 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -530,6 +610,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -583,6 +668,9 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -603,6 +691,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -656,6 +749,9 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -676,6 +772,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -732,6 +833,12 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
+				eventingrekttesting.NewSubscription(Subscription2Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription2UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -752,12 +859,22 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}, {
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription2UUID,
 								Destination:   "http://" + Subscription2URI,
 								Uid:           Subscription2UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_DiscardReply{},
+								Reference: &contract.Reference{
+									Name:      Subscription2Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription2UUID,
+								},
 							}},
 						},
 					},
@@ -1084,6 +1201,10 @@ func TestReconcileKind(t *testing.T) {
 				}),
 				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewLegacySSLSecret("ns-1", "secret-1"),
+				NewSSLSecret(system.Namespace(), "secret-1"),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -1122,6 +1243,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -1178,6 +1304,9 @@ func TestReconcileKind(t *testing.T) {
 				}),
 				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewLegacySASLSSLSecret("ns-1", "secret-1"),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -1219,6 +1348,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -1275,6 +1409,9 @@ func TestReconcileKind(t *testing.T) {
 				}),
 				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
 				NewLegacySASLSecret("ns-1", "secret-1"),
+				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
+					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
+				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -1313,6 +1450,11 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+								Reference: &contract.Reference{
+									Name:      Subscription1Name,
+									Namespace: ChannelNamespace,
+									Uuid:      Subscription1UUID,
+								},
 							}},
 						},
 					},
@@ -1627,9 +1769,10 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 				DispatcherLabel:             base.ChannelDispatcherLabel,
 				ReceiverLabel:               base.ChannelReceiverLabel,
 			},
-			Env:             env,
-			ConfigMapLister: listers.GetConfigMapLister(),
-			ServiceLister:   listers.GetServiceLister(),
+			Env:                env,
+			ConfigMapLister:    listers.GetConfigMapLister(),
+			ServiceLister:      listers.GetServiceLister(),
+			SubscriptionLister: listers.GetSubscriptionLister(),
 			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 				return 1, nil
 			},
diff --git a/control-plane/pkg/reconciler/channel/controller_test.go b/control-plane/pkg/reconciler/channel/controller_test.go
index f41e93c91..0dab8397d 100644
--- a/control-plane/pkg/reconciler/channel/controller_test.go
+++ b/control-plane/pkg/reconciler/channel/controller_test.go
@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake"
 	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
 	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
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 a94c0e919..54441899e 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
@@ -300,6 +300,71 @@ public class RecordDispatcherTest {
     assertNoDiscardedEventCount();
   }
 
+  @Test
+  public void failedEventsShouldBeEnhancedWithCustomHttpHeaders() {
+
+    final var subscriberSenderSendCalled = new AtomicBoolean(false);
+    final var dlsSenderSendCalled = new AtomicBoolean(false);
+    final RecordDispatcherListener receiver = offsetManagerMock();
+
+    int errorCode = 422;
+    String errorBody = "{ \"message\": \"bad bad things happened\" }";
+    String validErrorKey = "kne-testerror";
+    String invalidErrorKey = "something";
+    MultiMap headerMap = MultiMap.caseInsensitiveMultiMap().add(validErrorKey, "hello").add(invalidErrorKey,"nope");
+
+    final var dispatcherHandler = new RecordDispatcherImpl(
+      resourceContext,
+      value -> true,
+      new CloudEventSenderMock(
+        record -> {
+          subscriberSenderSendCalled.set(true);
+          return Future.failedFuture(new ResponseFailureException(makeHttpResponseWithHeaders(errorCode, errorBody, headerMap), ""));
+        }
+      ),
+      new CloudEventSenderMock(
+        record -> {
+          dlsSenderSendCalled.set(true);
+          return Future.succeededFuture();
+        }
+      ), new ResponseHandlerMock(),
+      receiver,
+      null,
+      registry
+    );
+    final var record = record();
+    dispatcherHandler.dispatch(record);
+
+    ArgumentCaptor<KafkaConsumerRecord<Object, CloudEvent>> captor = ArgumentCaptor.forClass(KafkaConsumerRecord.class);
+
+    assertTrue(subscriberSenderSendCalled.get());
+    assertTrue(dlsSenderSendCalled.get());
+    verify(receiver, times(1)).recordReceived(record);
+    verify(receiver, times(1)).successfullySentToDeadLetterSink(captor.capture());
+    verify(receiver, never()).successfullySentToSubscriber(any());
+    verify(receiver, never()).failedToSendToDeadLetterSink(any(), any());
+    verify(receiver, never()).recordDiscarded(any());
+
+    KafkaConsumerRecord<Object, CloudEvent> failedRecord = captor.getValue();
+    assertEquals(record.topic(), failedRecord.topic());
+    assertEquals(record.partition(), failedRecord.partition());
+    assertEquals(record.offset(), failedRecord.offset());
+    assertEquals(record.key(), failedRecord.key());
+    assertEquals(record.value().getId(), failedRecord.value().getId());
+    assertEquals(record.value().getAttributeNames(), failedRecord.value().getAttributeNames());
+    assertEquals(record.value().getData(), failedRecord.value().getData());
+    assertEquals("testdest", failedRecord.value().getExtension("knativeerrordest"));
+    assertEquals(String.valueOf(errorCode), failedRecord.value().getExtension("knativeerrorcode"));
+    assertEquals("hello", failedRecord.value().getExtension("testerror"));
+    assertThat(failedRecord.value().getExtension("something")).isNull();
+    assertEquals(Base64.getEncoder().encodeToString(errorBody.getBytes()), failedRecord.value().getExtension("knativeerrordata"));
+
+    assertEventDispatchLatency();
+    assertEventProcessingLatency();
+    assertEventCount();
+    assertNoDiscardedEventCount();
+  }
+
   @Test
   public void failedEventsShouldBeEnhancedWithErrorExtensionsPriorToSendingToDlsBodyTooLarge() {
 
@@ -362,11 +427,15 @@ public class RecordDispatcherTest {
   }
 
   private HttpResponseImpl<Buffer> makeHttpResponse(int statusCode, String body) {
+    return makeHttpResponseWithHeaders(statusCode, body, MultiMap.caseInsensitiveMultiMap());
+  }
+
+  private HttpResponseImpl<Buffer> makeHttpResponseWithHeaders(int statusCode, String body, MultiMap headers) {
     return new HttpResponseImpl<Buffer>(
       HttpVersion.HTTP_2,
       statusCode,
       "",
-      MultiMap.caseInsensitiveMultiMap(),
+      headers,
       MultiMap.caseInsensitiveMultiMap(),
       Collections.emptyList(),
       Buffer.buffer(body, "UTF-8"),
diff --git a/test/e2e_new/dls_extensions_test.go b/test/e2e_new/dls_extensions_test.go
index 893b14f6d..008c3d41e 100644
--- a/test/e2e_new/dls_extensions_test.go
+++ b/test/e2e_new/dls_extensions_test.go
@@ -59,6 +59,7 @@ func TestDeadLetterSinkExtensions(t *testing.T) {
 	env.Test(ctx, t, SubscriberReturnedErrorSmallData())
 	env.Test(ctx, t, SubscriberReturnedErrorLargeData())
 	env.Test(ctx, t, SubscriberReturnedHtmlWebpage())
+	env.Test(ctx, t, SubscriberReturnedCustomExtensionHeader())
 }
 
 func SubscriberUnreachable() *feature.Feature {
@@ -361,6 +362,73 @@ func SubscriberReturnedHtmlWebpage() *feature.Feature {
 	return f
 }
 
+func SubscriberReturnedCustomExtensionHeader() *feature.Feature {
+	f := feature.NewFeature()
+
+	sourceName := feature.MakeRandomK8sName("source")
+	sinkName := feature.MakeRandomK8sName("sink")
+	deadLetterSinkName := feature.MakeRandomK8sName("dls")
+	triggerName := feature.MakeRandomK8sName("trigger")
+	brokerName := feature.MakeRandomK8sName("broker")
+
+	ev := cetest.FullEvent()
+
+	f.Setup("install one partition configuration", single_partition_config.Install)
+	f.Setup("install broker", broker.Install(
+		brokerName,
+		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithConfig(single_partition_config.ConfigMapName),
+	))
+	f.Setup("broker is ready", broker.IsReady(brokerName))
+	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
+
+	errorData := `{ "message": "catastrophic failure" }`
+	f.Setup("install sink", eventshub.Install(
+		sinkName,
+		eventshub.StartReceiver,
+		eventshub.DropFirstN(1),
+		eventshub.DropEventsResponseCode(422),
+		eventshub.DropEventsResponseHeaders(map[string]string{"Kne-Test": "foo"}),
+		eventshub.DropEventsResponseBody(errorData),
+	))
+	f.Setup("install dead letter sink", eventshub.Install(
+		deadLetterSinkName,
+		eventshub.StartReceiver,
+	))
+	f.Setup("install trigger", trigger.Install(
+		triggerName,
+		brokerName,
+		trigger.WithSubscriber(svc.AsKReference(sinkName), ""),
+		trigger.WithDeadLetterSink(svc.AsKReference(deadLetterSinkName), ""),
+	))
+	f.Setup("trigger is ready", trigger.IsReady(triggerName))
+
+	f.Requirement("install source", eventshub.Install(
+		sourceName,
+		eventshub.StartSenderToResource(broker.GVR(), brokerName),
+		eventshub.InputEvent(ev),
+	))
+
+	f.Assert("knativeerrordest, knativeerrorcode, knativeerrordata and custom extension header added", assertEnhancedWithKnativeErrorExtensions(
+		deadLetterSinkName,
+		func(ctx context.Context) cetest.EventMatcher {
+			sinkAddress, _ := svc.Address(ctx, sinkName)
+			return cetest.HasExtension("knativeerrordest", sinkAddress.String())
+		},
+		func(ctx context.Context) cetest.EventMatcher {
+			return cetest.HasExtension("knativeerrorcode", "422")
+		},
+		func(ctx context.Context) cetest.EventMatcher {
+			return cetest.HasExtension("test", "foo")
+		},
+		func(ctx context.Context) cetest.EventMatcher {
+			return cetest.HasExtension("knativeerrordata", base64.StdEncoding.EncodeToString([]byte(errorData)))
+		},
+	))
+
+	return f
+}
+
 func assertEnhancedWithKnativeErrorExtensions(sinkName string, matcherfns ...func(ctx context.Context) cetest.EventMatcher) feature.StepFn {
 	return func(ctx context.Context, t feature.T) {
 		matchers := make([]cetest.EventMatcher, len(matcherfns))