Repository: knative-sandbox/eventing-kafka-broker
Pull request #1906 (2022-02-15 15:16:36)

Simulate dispatch latency during unit tests
The assertion for dispatch latency metrics is flaky since the
sender is a mock that doesn't do anything.
This ends up producing a dispatch latency of 0.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Simulate dispatch latency during unit tests

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1906#" 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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1906?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1906](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1906?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (d5a0c37) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/3f5dce3e5174cc7d74aae996e14b6d4a671f8c12?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (3f5dce3) will **not change** coverage.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1906/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1906?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@            Coverage Diff            @@
##               main    #1906   +/-   ##
=========================================
  Coverage     67.63%   67.63%           
  Complexity      632      632           
=========================================
  Files           138      138           
  Lines          5840     5840           
  Branches        179      179           
=========================================
  Hits           3950     3950           
  Misses         1575     1575           
  Partials        315      315           
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.31% <ø> (ø)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.


------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1906?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1906?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [3f5dce3...d5a0c37](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1906?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-prow-robot: @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
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl | d5a0c37cc9f56a4b0c67f561e55f957fe89ab612 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1906/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl/1493605172658573312) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1906). [Your PR dashboard](https://gubernator.knative.dev/pr/pierDipi).

<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-robot: @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
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl | d5a0c37cc9f56a4b0c67f561e55f957fe89ab612 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1906/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl/1493605172658573312) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl`
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl | d5a0c37cc9f56a4b0c67f561e55f957fe89ab612 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1906/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl/1493605172692127744) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1906). [Your PR dashboard](https://gubernator.knative.dev/pr/pierDipi).

<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: /test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl
/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-sasl-ssl

Patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
index d28346249..7beb3bd2d 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
@@ -98,22 +98,28 @@ type ConsumerSpec struct {
 	CloudEventOverrides *duckv1.CloudEventOverrides `json:"ceOverrides,omitempty"`
 
 	// VReplicas is the number of virtual replicas for a consumer.
-	VReplicas *int32
+	VReplicas *int32 `json:"vReplicas"`
 
 	// PodBind represents a reference to the pod in which the consumer should be placed.
 	PodBind *PodBind `json:"podBind"`
 }
 
 type ReplyStrategy struct {
-	TopicReply *TopicReply
-	URLReply   *DestinationReply
-	NoReply    *struct{}
+	TopicReply *TopicReply       `json:"topicReply,omitempty"`
+	URLReply   *DestinationReply `json:"URLReply,omitempty"`
+	NoReply    *NoReply          `json:"noReply,omitempty"`
+}
+
+type NoReply struct {
+	Enabled bool `json:"enabled"`
 }
 
 type TopicReply struct {
+	Enabled bool `json:"enabled"`
 }
 
 type DestinationReply struct {
+	Enabled     bool `json:"enabled"`
 	Destination duckv1.Destination
 }
 
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go
index f8dd77591..db02d7e4b 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go
@@ -45,6 +45,7 @@ func (cs *ConsumerSpec) Validate(ctx context.Context) *apis.FieldError {
 		cs.Subscriber.Validate(ctx).ViaField("subscriber"),
 		cs.PodBind.Validate(ctx).ViaField("podBind"),
 		cs.CloudEventOverrides.Validate(ctx).ViaField("ceOverrides"),
+		cs.Reply.Validate(ctx).ViaField("reply"),
 	)
 	return err
 }
@@ -73,6 +74,27 @@ func (f *Filters) Validate(ctx context.Context) *apis.FieldError {
 	return nil
 }
 
+func (in *ReplyStrategy) Validate(ctx context.Context) *apis.FieldError {
+	if in == nil {
+		return nil
+	}
+	setCounter := 0
+	if in.TopicReply != nil && in.TopicReply.Enabled {
+		setCounter++
+	}
+	if in.URLReply != nil && in.URLReply.Enabled {
+		setCounter++
+	}
+	if in.NoReply != nil && in.NoReply.Enabled {
+		setCounter++
+	}
+
+	if setCounter > 1 {
+		return apis.ErrMultipleOneOf("topicReply", "URLReply", "NoReply")
+	}
+	return nil
+}
+
 func (p *PodBind) Validate(ctx context.Context) *apis.FieldError {
 	if p == nil {
 		return apis.ErrMissingField("")
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
index 42bfddaa6..2498b1e2d 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
@@ -451,6 +451,22 @@ func (in *Filters) DeepCopy() *Filters {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *NoReply) DeepCopyInto(out *NoReply) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new NoReply.
+func (in *NoReply) DeepCopy() *NoReply {
+	if in == nil {
+		return nil
+	}
+	out := new(NoReply)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *PodBind) DeepCopyInto(out *PodBind) {
 	*out = *in
@@ -482,7 +498,7 @@ func (in *ReplyStrategy) DeepCopyInto(out *ReplyStrategy) {
 	}
 	if in.NoReply != nil {
 		in, out := &in.NoReply, &out.NoReply
-		*out = new(struct{})
+		*out = new(NoReply)
 		**out = **in
 	}
 	return
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 57fb269c2..d63fee73b 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -430,7 +430,7 @@ func (r *Reconciler) reconcileSubscribers(ctx context.Context, kafkaClient saram
 				UID:                s.UID,
 				ObservedGeneration: s.Generation,
 				Ready:              corev1.ConditionFalse,
-				Message:            fmt.Sprint("Subscription not ready", err),
+				Message:            fmt.Sprintf("Subscription not ready: %v", err),
 			})
 			globalErr = multierr.Append(globalErr, err)
 		} else {
@@ -479,8 +479,13 @@ func (r *Reconciler) reconcileInitialOffset(ctx context.Context, channel *messag
 }
 
 func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (*contract.Egress, error) {
+	subscriberURI := subscriber.SubscriberURI.String()
+	if subscriberURI == "" {
+		return nil, fmt.Errorf("failed to resolve Subscription.Spec.Subscriber: empty subscriber URI")
+	}
+
 	egress := &contract.Egress{
-		Destination:   subscriber.SubscriberURI.String(),
+		Destination:   subscriberURI,
 		ConsumerGroup: consumerGroup(channel, subscriber),
 		DeliveryOrder: DefaultDeliveryOrder,
 		Uid:           string(subscriber.UID),
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index d225b5966..b4b58c264 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -252,11 +252,11 @@ func (r Reconciler) reconcileReplyStrategy(ctx context.Context, c *kafkainternal
 	if c.Spec.Reply == nil {
 		return nil
 	}
-	if c.Spec.Reply.NoReply != nil {
+	if c.Spec.Reply.NoReply != nil && c.Spec.Reply.NoReply.Enabled {
 		egress.ReplyStrategy = &contract.Egress_DiscardReply{}
 		return nil
 	}
-	if c.Spec.Reply.URLReply != nil {
+	if c.Spec.Reply.URLReply != nil && c.Spec.Reply.URLReply.Enabled {
 		destination, err := r.Resolver.URIFromDestinationV1(ctx, c.Spec.Reply.URLReply.Destination, c)
 		if err != nil {
 			return fmt.Errorf("failed to resolve reply destination: %w", err)
@@ -266,7 +266,7 @@ func (r Reconciler) reconcileReplyStrategy(ctx context.Context, c *kafkainternal
 		}
 		return nil
 	}
-	if c.Spec.Reply.TopicReply != nil {
+	if c.Spec.Reply.TopicReply != nil && c.Spec.Reply.TopicReply.Enabled {
 		egress.ReplyStrategy = &contract.Egress_ReplyToOriginalTopic{}
 		return nil
 	}
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2.go b/control-plane/pkg/reconciler/source/v2/sourcev2.go
index 88e1e1e20..6558ccf0a 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2.go
@@ -114,6 +114,7 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 						Ordering: DefaultDeliveryOrder,
 					},
 					Subscriber: ks.Spec.Sink,
+					Reply:      &internalscg.ReplyStrategy{NoReply: &internalscg.NoReply{Enabled: true}},
 				},
 			},
 			Replicas: ks.Spec.Consumers,
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
index eade6ff7b..0e19e9b78 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
@@ -140,6 +140,7 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, trigger *eventin
 						Filters: trigger.Spec.Filters,
 					},
 					Subscriber: trigger.Spec.Subscriber,
+					Reply:      &internalscg.ReplyStrategy{TopicReply: &internalscg.TopicReply{Enabled: true}},
 				},
 			},
 		},
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 6096322d8..deca8a008 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -154,7 +154,7 @@ Lists of 158 third-party dependencies.
      (Apache-2.0) Scala Compiler (org.scala-lang:scala-reflect:2.12.13 - https://www.scala-lang.org/)
      (Apache-2.0) scala-collection-compat (org.scala-lang.modules:scala-collection-compat_2.12:2.3.0 - http://www.scala-lang.org/)
      (Apache-2.0) scala-java8-compat (org.scala-lang.modules:scala-java8-compat_2.12:0.9.1 - http://www.scala-lang.org/)
-     (MIT License) SLF4J API Module (org.slf4j:slf4j-api:1.7.35 - http://www.slf4j.org)
-     (MIT License) SLF4J NOP Binding (org.slf4j:slf4j-nop:1.7.35 - http://www.slf4j.org)
+     (MIT License) SLF4J API Module (org.slf4j:slf4j-api:1.7.36 - http://www.slf4j.org)
+     (MIT License) SLF4J NOP Binding (org.slf4j:slf4j-nop:1.7.36 - http://www.slf4j.org)
      (Apache-2.0) snappy-java (org.xerial.snappy:snappy-java:1.1.8.1 - https://github.com/xerial/snappy-java)
      (Apache License, Version 2.0) SnakeYAML (org.yaml:snakeyaml:1.28 - http://www.snakeyaml.org)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index a80f071ad..0c2d2f173 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -47,7 +47,7 @@
     <opentelemetry.version>1.11.0</opentelemetry.version>
     <jackson.version>2.13.1</jackson.version>
     <protobuf.version>3.19.4</protobuf.version>
-    <slf4j.version>1.7.35</slf4j.version>
+    <slf4j.version>1.7.36</slf4j.version>
     <ch.qos.logback.version>1.2.10</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>7.0.1</net.logstash.logback.encoder.version>
     <assertj.version>3.22.0</assertj.version>


Test patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
index 3e00b64c7..504c7e9e8 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
@@ -118,6 +118,45 @@ func TestConsumerGroup_Validate(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "invalid multiple reply strategies (topic, destination)",
+			ctx:  context.Background(),
+			given: &ConsumerGroup{
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Reply: &ReplyStrategy{
+								TopicReply: &TopicReply{Enabled: true},
+								URLReply:   &DestinationReply{Enabled: true},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
 		{
 			name: "invalid no pod name",
 			ctx:  context.Background(),
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index c5e71fdac..a888eee34 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -353,6 +353,79 @@ func TestReconcileKind(t *testing.T) {
 				testProber: probertesting.MockProber(prober.StatusUnknown),
 			},
 		},
+		{
+			Name: "Reconciled failed - with single fresh subscriber without URI",
+			Objects: []runtime.Object{
+				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber, WithNoSubscriberURI))),
+				NewService(),
+				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, nil),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+							Egresses: []*contract.Egress{},
+						},
+					},
+				}),
+				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",
+				}),
+			},
+			WantErr:                 true,
+			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,
+						//StatusInitialOffsetsCommitted,
+						WithSubscribers(Subscriber1(WithFreshSubscriber, WithNoSubscriberURI)),
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					"failed to resolve subscriber config: failed to resolve Subscription.Spec.Subscriber: empty subscriber URI",
+				),
+			},
+		},
 		{
 			Name: "Reconciled normal - with single fresh subscriber - no auth",
 			Objects: []runtime.Object{
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
index eec524aeb..070abf219 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
@@ -70,6 +70,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 				),
@@ -107,6 +108,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 						ConsumerCloudEventOverrides(&duckv1.CloudEventOverrides{
 							Extensions: map[string]string{"a": "foo", "b": "foo"},
 						}),
@@ -144,6 +146,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 					ConsumerGroupReady,
@@ -166,6 +169,7 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerAuth(NewConsumerSpecAuth()),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 							ConsumerSubscriber(NewSourceSink2Reference()),
+							ConsumerReply(ConsumerNoReply()),
 						)),
 						ConsumerGroupReplicas(1),
 						ConsumerGroupReady,
@@ -212,6 +216,7 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerAuth(NewConsumerSpecAuth()),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 							ConsumerSubscriber(NewSourceSinkReference()),
+							ConsumerReply(ConsumerNoReply()),
 						)),
 						ConsumerGroupReplicas(1),
 						ConsumerGroupReady,
@@ -257,6 +262,7 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerAuth(NewConsumerSpecAuth()),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 							ConsumerSubscriber(NewSourceSinkReference()),
+							ConsumerReply(ConsumerNoReply()),
 						)),
 						ConsumerGroupReplicas(1),
 					),
@@ -289,6 +295,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 					ConsumerGroupReady,
@@ -322,6 +329,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 				),
@@ -355,6 +363,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 					WithConsumerGroupFailed("failed", "failed"),
@@ -389,6 +398,7 @@ func TestReconcileKind(t *testing.T) {
 						ConsumerAuth(NewConsumerSpecAuth()),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
 					)),
 					ConsumerGroupReplicas(1),
 					ConsumerGroupReplicasStatus(1),
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
index cdfe2c636..a1147647b 100644
--- a/control-plane/pkg/reconciler/testing/objects_channel.go
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -261,6 +261,18 @@ func WithFreshSubscriber(sub *SubscriberInfo) {
 	sub.status = nil
 }
 
+func WithNoSubscriberURI(sub *SubscriberInfo) {
+	sub.spec.SubscriberURI = nil
+	if sub.status == nil {
+		sub.status = &eventingduckv1.SubscriberStatus{
+			UID:                sub.spec.UID,
+			ObservedGeneration: sub.spec.Generation,
+		}
+	}
+	sub.status.Ready = "False"
+	sub.status.Message = "Subscription not ready: failed to resolve subscriber config: failed to resolve Subscription.Spec.Subscriber: empty subscriber URI"
+}
+
 func WithUnreadySubscriber(sub *SubscriberInfo) {
 	sub.status.Ready = "False"
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_consumer.go b/control-plane/pkg/reconciler/testing/objects_consumer.go
index bb2a6ca3b..1a99d7f2f 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -226,3 +226,17 @@ func ConsumerReady() ConsumerOption {
 		}
 	}
 }
+
+func ConsumerReply(s *kafkainternals.ReplyStrategy) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.Reply = s
+	}
+}
+
+func ConsumerNoReply() *kafkainternals.ReplyStrategy {
+	return &kafkainternals.ReplyStrategy{NoReply: &kafkainternals.NoReply{Enabled: true}}
+}
+
+func ConsumerTopicReply() *kafkainternals.ReplyStrategy {
+	return &kafkainternals.ReplyStrategy{TopicReply: &kafkainternals.TopicReply{Enabled: true}}
+}
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
index 6e6fe7e24..f723e0743 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
@@ -91,6 +91,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 				),
 			},
@@ -128,6 +129,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 				),
 			},
@@ -166,6 +168,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 				),
 			},
@@ -244,6 +247,7 @@ func TestReconcileKind(t *testing.T) {
 							),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 							ConsumerFilters(NewConsumerSpecFilters()),
+							ConsumerReply(ConsumerTopicReply()),
 						)),
 						ConsumerGroupReady,
 					),
@@ -291,6 +295,7 @@ func TestReconcileKind(t *testing.T) {
 							),
 							ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 							ConsumerFilters(NewConsumerSpecFilters()),
+							ConsumerReply(ConsumerTopicReply()),
 						)),
 					),
 				},
@@ -326,6 +331,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 					ConsumerGroupReady,
 				),
@@ -362,6 +368,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 				),
 			},
@@ -397,6 +404,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 					WithConsumerGroupFailed("failed", "failed"),
 				),
@@ -433,6 +441,7 @@ func TestReconcileKind(t *testing.T) {
 						),
 						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
 						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
 					)),
 					WithDeadLetterSinkURI(url.String()),
 				),
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 a297cf662..9f7099d61 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
@@ -118,6 +118,7 @@ public class RecordDispatcherTest {
       new CloudEventSenderMock(
         record -> {
           sendCalled.set(true);
+          simulateLatency(); // Simulate dispatch latency
           return Future.succeededFuture();
         }
       ),
@@ -160,6 +161,7 @@ public class RecordDispatcherTest {
       new CloudEventSenderMock(
         record -> {
           subscriberSenderSendCalled.set(true);
+          simulateLatency(); // Simulate dispatch latency
           return Future.failedFuture("");
         }
       ),
@@ -201,6 +203,7 @@ public class RecordDispatcherTest {
       value -> true, new CloudEventSenderMock(
       record -> {
         subscriberSenderSendCalled.set(true);
+        simulateLatency(); // Simulate dispatch latency
         return Future.failedFuture("");
       }
     ),
@@ -242,6 +245,7 @@ public class RecordDispatcherTest {
       new CloudEventSenderMock(
         record -> {
           subscriberSenderSendCalled.set(true);
+          simulateLatency(); // Simulate dispatch latency
           return Future.failedFuture("");
         }
       ),
@@ -381,4 +385,11 @@ public class RecordDispatcherTest {
     );
   }
 
+  private void simulateLatency() {
+    try {
+      Thread.sleep(100);
+    } catch (InterruptedException ignored) {
+    }
+  }
+
 }