Repository: knative-sandbox/eventing-kafka-broker
Pull request #2019 (2022-03-18 10:37:49)

Propagate consumer group status in Trigger and fallback to Broker delivery spec
As per title.

/kind cleanup

pierDipi: /assign @aavarghese @matzew 
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/2019#" 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-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/v2/triggerv2.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2019/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1504769037261869056/artifacts/line-cov.html#file0) | 90.0% | 89.1% | -0.9

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/v2/triggerv2.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2019/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1504772027397967872/artifacts/line-cov.html#file0) | 90.0% | 89.1% | -0.9

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2019](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (2e64fb1) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/84ab89b7eba5e130b2b4d576a042e1487f93a53e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (84ab89b) will **decrease** coverage by `6.96%`.
> The diff coverage is `90.00%`.

> :exclamation: Current head 2e64fb1 differs from pull request most recent head 6b742b0. Consider uploading reports for the commit 6b742b0 to get more accurate results

```diff
@@             Coverage Diff              @@
##               main    #2019      +/-   ##
============================================
- Coverage     67.25%   60.29%   -6.97%     
============================================
  Files           141       69      -72     
  Lines          6707     4536    -2171     
  Branches        187        0     -187     
============================================
- Hits           4511     2735    -1776     
+ Misses         1820     1524     -296     
+ Partials        376      277      -99     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

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.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ntrol-plane/pkg/reconciler/trigger/v2/triggerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3YyL3RyaWdnZXJ2Mi5nbw==) | `87.50% <90.00%> (-1.39%)` | :arrow_down: |
| [...ting/kafka/broker/dispatcher/CloudEventSender.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvQ2xvdWRFdmVudFNlbmRlci5qYXZh) | | |
| [...eceiver/impl/IngressProducerReconcilableStore.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL0luZ3Jlc3NQcm9kdWNlclJlY29uY2lsYWJsZVN0b3JlLmphdmE=) | | |
| [.../broker/receiver/main/ReceiverVerticleFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL1JlY2VpdmVyVmVydGljbGVGYWN0b3J5LmphdmE=) | | |
| [...r/dispatcher/main/ConsumerVerticleFactoryImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lclZlcnRpY2xlRmFjdG9yeUltcGwuamF2YQ==) | | |
| [...atcher/impl/consumer/KafkaConsumerRecordUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9LYWZrYUNvbnN1bWVyUmVjb3JkVXRpbHMuamF2YQ==) | | |
| [...atcher/impl/filter/subscriptionsapi/AllFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvc3Vic2NyaXB0aW9uc2FwaS9BbGxGaWx0ZXIuamF2YQ==) | | |
| [...ver/impl/handler/ControlPlaneProbeRequestUtil.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvQ29udHJvbFBsYW5lUHJvYmVSZXF1ZXN0VXRpbC5qYXZh) | | |
| [...ng/kafka/broker/dispatcher/main/DispatcherEnv.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9EaXNwYXRjaGVyRW52LmphdmE=) | | |
| [...broker/dispatcher/impl/consumer/OffsetManager.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9PZmZzZXRNYW5hZ2VyLmphdmE=) | | |
| ... and [63 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019?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/2019?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [84ab89b...6b742b0](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2019?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).


Patch:
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index 0cb720b02..f5ac60934 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -57,11 +57,11 @@ type Reconciler struct {
 }
 
 var (
-	_ consumer.Interface = Reconciler{}
-	_ consumer.Finalizer = Reconciler{}
+	_ consumer.Interface = &Reconciler{}
+	_ consumer.Finalizer = &Reconciler{}
 )
 
-func (r Reconciler) ReconcileKind(ctx context.Context, c *kafkainternals.Consumer) reconciler.Event {
+func (r *Reconciler) ReconcileKind(ctx context.Context, c *kafkainternals.Consumer) reconciler.Event {
 	logger := logging.FromContext(ctx).Desugar()
 
 	resourceCt, err := r.reconcileContractResource(ctx, c)
@@ -88,7 +88,7 @@ func (r Reconciler) ReconcileKind(ctx context.Context, c *kafkainternals.Consume
 	return nil
 }
 
-func (r Reconciler) FinalizeKind(ctx context.Context, c *kafkainternals.Consumer) reconciler.Event {
+func (r *Reconciler) FinalizeKind(ctx context.Context, c *kafkainternals.Consumer) reconciler.Event {
 
 	logger := logging.FromContext(ctx).Desugar()
 
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2.go b/control-plane/pkg/reconciler/source/v2/sourcev2.go
index c588ee2cc..2cede37c4 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2.go
@@ -24,7 +24,9 @@ import (
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
@@ -61,36 +63,16 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource)
 		ks.GetConditionSet().Manage(&ks.Status).MarkFalse(KafkaConditionConsumerGroup, "failed to reconcile consumer group", err.Error())
 		return err
 	}
-	if cg.IsReady() {
-		ks.GetConditionSet().Manage(&ks.Status).MarkTrue(KafkaConditionConsumerGroup)
-	} else {
-		topLevelCondition := cg.GetConditionSet().Manage(cg.GetStatus()).GetTopLevelCondition()
-		if topLevelCondition == nil {
-			ks.GetConditionSet().Manage(&ks.Status).MarkUnknown(
-				KafkaConditionConsumerGroup,
-				"failed to reconcile consumer group",
-				"consumer group not ready",
-			)
-		} else {
-			ks.GetConditionSet().Manage(&ks.Status).MarkFalse(
-				KafkaConditionConsumerGroup,
-				topLevelCondition.Reason,
-				topLevelCondition.Message,
-			)
-		}
-	}
 
-	ks.Status.MarkSink(cg.Status.SubscriberURI)
-	ks.Status.Placeable = cg.Status.Placeable
-	if cg.Status.Replicas != nil {
-		ks.Status.Consumers = *cg.Status.Replicas
-	}
+	propagateConsumerGroupStatus(cg, ks)
 
 	return nil
 }
 
 func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.KafkaSource) (*internalscg.ConsumerGroup, error) {
 
+	backoffPolicy := eventingduck.BackoffPolicyExponential
+
 	expectedCg := &internalscg.ConsumerGroup{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      string(ks.UID),
@@ -119,6 +101,12 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 						NetSpec: &ks.Spec.KafkaAuthSpec.Net,
 					},
 					Delivery: &internalscg.DeliverySpec{
+						DeliverySpec: &eventingduck.DeliverySpec{
+							Retry:         pointer.Int32(10),
+							BackoffPolicy: &backoffPolicy,
+							BackoffDelay:  pointer.String("PT10S"),
+							Timeout:       pointer.String("PT600S"),
+						},
 						Ordering: DefaultDeliveryOrder,
 					},
 					Subscriber: ks.Spec.Sink,
@@ -163,3 +151,29 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 
 	return cg, nil
 }
+
+func propagateConsumerGroupStatus(cg *internalscg.ConsumerGroup, ks *sources.KafkaSource) {
+	if cg.IsReady() {
+		ks.GetConditionSet().Manage(&ks.Status).MarkTrue(KafkaConditionConsumerGroup)
+	} else {
+		topLevelCondition := cg.GetConditionSet().Manage(cg.GetStatus()).GetTopLevelCondition()
+		if topLevelCondition == nil {
+			ks.GetConditionSet().Manage(&ks.Status).MarkUnknown(
+				KafkaConditionConsumerGroup,
+				"failed to reconcile consumer group",
+				"consumer group is not ready",
+			)
+		} else {
+			ks.GetConditionSet().Manage(&ks.Status).MarkFalse(
+				KafkaConditionConsumerGroup,
+				topLevelCondition.Reason,
+				topLevelCondition.Message,
+			)
+		}
+	}
+	ks.Status.MarkSink(cg.Status.SubscriberURI)
+	ks.Status.Placeable = cg.Status.Placeable
+	if cg.Status.Replicas != nil {
+		ks.Status.Consumers = *cg.Status.Replicas
+	}
+}
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
index 90ab36c6b..c2a95da3c 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	eventingclientset "knative.dev/eventing/pkg/client/clientset/versioned"
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
@@ -83,31 +84,17 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, trigger *eventing.Trigge
 		return nil
 	}
 
-	cg, err := r.reconcileConsumerGroup(ctx, trigger)
+	cg, err := r.reconcileConsumerGroup(ctx, broker, trigger)
 	if err != nil {
 		trigger.Status.MarkDependencyFailed("failed to reconcile consumer group", err.Error())
 		return err
 	}
-	if cg.IsReady() {
-		trigger.Status.MarkDependencySucceeded()
-	} else {
-		topLevelCondition := cg.GetConditionSet().Manage(cg.GetStatus()).GetTopLevelCondition()
-		if topLevelCondition == nil {
-			trigger.Status.MarkDependencyUnknown("failed to reconcile consumer group", "consumer group not ready")
-		} else {
-			trigger.Status.MarkDependencyFailed(topLevelCondition.Reason, topLevelCondition.Message)
-		}
-	}
-	trigger.Status.SubscriberURI = cg.Status.SubscriberURI
-	trigger.Status.MarkSubscriberResolvedSucceeded()
-
-	trigger.Status.DeadLetterSinkURI = cg.Status.DeadLetterSinkURI
-	trigger.Status.MarkDeadLetterSinkResolvedSucceeded()
+	propagateConsumerGroupStatus(cg, trigger)
 
 	return nil
 }
 
-func (r Reconciler) reconcileConsumerGroup(ctx context.Context, trigger *eventing.Trigger) (*internalscg.ConsumerGroup, error) {
+func (r Reconciler) reconcileConsumerGroup(ctx context.Context, broker *eventing.Broker, trigger *eventing.Trigger) (*internalscg.ConsumerGroup, error) {
 
 	var deliveryOrdering = internals.Unordered
 	var err error
@@ -141,8 +128,9 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, trigger *eventin
 					Topics:  []string{},                                                                               //todo get topics from broker resource
 					Configs: internalscg.ConsumerConfigs{Configs: map[string]string{"group.id": string(trigger.UID)}}, //todo get bootstrap.servers from broker resource
 					Delivery: &internalscg.DeliverySpec{
-						DeliverySpec: trigger.Spec.Delivery,
-						Ordering:     deliveryOrdering},
+						DeliverySpec: deliverySpec(broker, trigger),
+						Ordering:     deliveryOrdering,
+					},
 					Filters: &internalscg.Filters{
 						Filter:  trigger.Spec.Filter,
 						Filters: trigger.Spec.Filters,
@@ -187,6 +175,32 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, trigger *eventin
 	return cg, nil
 }
 
+func deliverySpec(broker *eventing.Broker, trigger *eventing.Trigger) *eventingduck.DeliverySpec {
+	// TOOD(pierDipi) use `Merge` in https://github.com/knative/eventing/pull/6277/files
+	if trigger.Spec.Delivery != nil {
+		return trigger.Spec.Delivery
+	}
+	return broker.Spec.Delivery
+}
+
+func propagateConsumerGroupStatus(cg *internalscg.ConsumerGroup, trigger *eventing.Trigger) {
+	if cg.IsReady() {
+		trigger.Status.MarkDependencySucceeded()
+	} else {
+		topLevelCondition := cg.GetConditionSet().Manage(cg.GetStatus()).GetTopLevelCondition()
+		if topLevelCondition == nil {
+			trigger.Status.MarkDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready")
+		} else {
+			trigger.Status.MarkDependencyFailed(topLevelCondition.Reason, topLevelCondition.Message)
+		}
+	}
+	trigger.Status.SubscriberURI = cg.Status.SubscriberURI
+	trigger.Status.MarkSubscriberResolvedSucceeded()
+
+	trigger.Status.DeadLetterSinkURI = cg.Status.DeadLetterSinkURI
+	trigger.Status.MarkDeadLetterSinkResolvedSucceeded()
+}
+
 func isKnativeKafkaBroker(broker *eventing.Broker) (bool, string) {
 	brokerClass := broker.GetAnnotations()[eventing.BrokerClassAnnotationKey]
 	return brokerClass == kafka.BrokerClass, brokerClass


Test patch:
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
index a80d86449..75796af45 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
@@ -23,6 +23,7 @@ import (
 
 	"k8s.io/apimachinery/pkg/runtime"
 	clientgotesting "k8s.io/client-go/testing"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
 
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
@@ -69,7 +70,15 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerBootstrapServersConfig(SourceBootstrapServers),
 						),
 						ConsumerAuth(NewConsumerSpecAuth()),
-						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
 						ConsumerSubscriber(NewSourceSinkReference()),
 						ConsumerReply(ConsumerNoReply()),
 					)),
@@ -108,7 +117,15 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerBootstrapServersConfig(SourceBootstrapServers),
 						),
 						ConsumerAuth(NewConsumerSpecAuth()),
-						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
 						ConsumerSubscriber(NewSourceSinkReference()),
 						ConsumerReply(ConsumerNoReply()),
 						ConsumerCloudEventOverrides(&duckv1.CloudEventOverrides{
@@ -147,7 +164,15 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerBootstrapServersConfig(SourceBootstrapServers),
 						),
 						ConsumerAuth(NewConsumerSpecAuth()),
-						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
 						ConsumerSubscriber(NewSourceSinkReference()),
 						ConsumerReply(ConsumerNoReply()),
 					)),
@@ -171,7 +196,15 @@ func TestReconcileKind(t *testing.T) {
 								ConsumerBootstrapServersConfig(SourceBootstrapServers),
 							),
 							ConsumerAuth(NewConsumerSpecAuth()),
-							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+							ConsumerDelivery(
+								NewConsumerSpecDelivery(
+									internals.Ordered,
+									NewConsumerTimeout("PT600S"),
+									NewConsumerRetry(10),
+									NewConsumerBackoffDelay("PT10S"),
+									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								),
+							),
 							ConsumerSubscriber(NewSourceSink2Reference()),
 							ConsumerReply(ConsumerNoReply()),
 						)),
@@ -220,7 +253,15 @@ func TestReconcileKind(t *testing.T) {
 								ConsumerBootstrapServersConfig(SourceBootstrapServers),
 							),
 							ConsumerAuth(NewConsumerSpecAuth()),
-							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+							ConsumerDelivery(
+								NewConsumerSpecDelivery(
+									internals.Ordered,
+									NewConsumerTimeout("PT600S"),
+									NewConsumerRetry(10),
+									NewConsumerBackoffDelay("PT10S"),
+									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								),
+							),
 							ConsumerSubscriber(NewSourceSinkReference()),
 							ConsumerReply(ConsumerNoReply()),
 						)),
@@ -268,7 +309,15 @@ func TestReconcileKind(t *testing.T) {
 								ConsumerBootstrapServersConfig(SourceBootstrapServers),
 							),
 							ConsumerAuth(NewConsumerSpecAuth()),
-							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+							ConsumerDelivery(
+								NewConsumerSpecDelivery(
+									internals.Ordered,
+									NewConsumerTimeout("PT600S"),
+									NewConsumerRetry(10),
+									NewConsumerBackoffDelay("PT10S"),
+									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								),
+							),
 							ConsumerSubscriber(NewSourceSinkReference()),
 							ConsumerReply(ConsumerNoReply()),
 						)),
@@ -302,7 +351,15 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerBootstrapServersConfig(SourceBootstrapServers),
 						),
 						ConsumerAuth(NewConsumerSpecAuth()),
-						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
 						ConsumerSubscriber(NewSourceSinkReference()),
 						ConsumerReply(ConsumerNoReply()),
 					)),
@@ -337,7 +394,15 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerBootstrapServersConfig(SourceBootstrapServers),
 						),
 						ConsumerAuth(NewConsumerSpecAuth()),
-						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
 						ConsumerSubscriber(NewSourceSinkReference()),
 						ConsumerReply(ConsumerNoReply()),
 					)),
@@ -372,7 +437,15 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerBootstrapServersConfig(SourceBootstrapServers),
 						),
 						ConsumerAuth(NewConsumerSpecAuth()),
-						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
 						ConsumerSubscriber(NewSourceSinkReference()),
 						ConsumerReply(ConsumerNoReply()),
 					)),
@@ -408,7 +481,15 @@ func TestReconcileKind(t *testing.T) {
 							ConsumerBootstrapServersConfig(SourceBootstrapServers),
 						),
 						ConsumerAuth(NewConsumerSpecAuth()),
-						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
 						ConsumerSubscriber(NewSourceSinkReference()),
 						ConsumerReply(ConsumerNoReply()),
 					)),
@@ -467,7 +548,7 @@ func StatusSourceConsumerGroupFailed(reason string, msg string) KRShapedOption {
 func StatusSourceConsumerGroupUnknown() KRShapedOption {
 	return func(obj duckv1.KRShaped) {
 		ks := obj.(*sources.KafkaSource)
-		ks.GetConditionSet().Manage(ks.GetStatus()).MarkUnknown(KafkaConditionConsumerGroup, "failed to reconcile consumer group", "consumer group not ready")
+		ks.GetConditionSet().Manage(ks.GetStatus()).MarkUnknown(KafkaConditionConsumerGroup, "failed to reconcile consumer group", "consumer group is not ready")
 	}
 }
 
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
index 19e7102e5..417d586a8 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
@@ -21,6 +21,9 @@ import (
 	"fmt"
 	"testing"
 
+	"k8s.io/utils/pointer"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 
@@ -62,6 +65,8 @@ var (
 		Host:   "localhost",
 		Path:   "/path",
 	}
+
+	exponential = eventingduck.BackoffPolicyExponential
 )
 
 func TestReconcileKind(t *testing.T) {
@@ -102,7 +107,52 @@ func TestReconcileKind(t *testing.T) {
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(),
-						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - fallback to broker delivery",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+					WithRetry(pointer.Int32Ptr(10), &exponential, pointer.StringPtr("PT2S")),
+				),
+				newTrigger(),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupMetaLabels(OwnerAsTriggerLabel),
+					WithConsumerGroupLabels(ConsumerTriggerLabel),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(
+							internals.Unordered,
+							NewConsumerRetry(10),
+							NewConsumerBackoffDelay("PT2S"),
+							NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+						)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
+					)),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
 						withDeadLetterSinkURI(""),
 					),
 				},
@@ -141,7 +191,7 @@ func TestReconcileKind(t *testing.T) {
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(),
-						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
 						withDeadLetterSinkURI(""),
 						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Ordered)),
 					),
@@ -181,7 +231,7 @@ func TestReconcileKind(t *testing.T) {
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(),
-						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
 						withDeadLetterSinkURI(""),
 						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Unordered)),
 					),
@@ -313,7 +363,7 @@ func TestReconcileKind(t *testing.T) {
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(),
-						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
 						withDeadLetterSinkURI(""),
 					),
 				},
@@ -388,7 +438,7 @@ func TestReconcileKind(t *testing.T) {
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(),
-						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
 						withDeadLetterSinkURI(""),
 					),
 				},
@@ -464,7 +514,7 @@ func TestReconcileKind(t *testing.T) {
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(),
-						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
 						withDeadLetterSinkURI(url.String()),
 					),
 				},