Repository: knative-sandbox/eventing-kafka-broker
Pull request #2006 (2022-03-16 09:37:48)

Config retries for KafkaSource scheduler based
As per title.

These values are the same we use in the existing
KafkaSource in this repository

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/2006#" 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":[]} -->
pierDipi: /assign @aavarghese 
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/source/v2/sourcev2.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2006/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1504029158873763840/artifacts/line-cov.html#file1) | 84.4% | 84.8% | 0.5

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2006?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2006](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2006?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (ee42acd) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/beea3153b9ad456f9df6ef8aafbc0da9943b885c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (beea315) will **increase** coverage by `0.06%`.
> The diff coverage is `100.00%`.

```diff
@@             Coverage Diff              @@
##               main    #2006      +/-   ##
============================================
+ Coverage     67.27%   67.33%   +0.06%     
  Complexity      639      639              
============================================
  Files           141      141              
  Lines          6692     6699       +7     
  Branches        186      186              
============================================
+ Hits           4502     4511       +9     
+ Misses         1816     1815       -1     
+ Partials        374      373       -1     
```

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

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/2006?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/consumer/consumer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2006/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb25zdW1lci5nbw==) | `50.00% <ø> (ø)` | |
| [control-plane/pkg/reconciler/source/v2/sourcev2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2006/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvdjIvc291cmNldjIuZ28=) | `89.77% <100.00%> (+0.88%)` | :arrow_up: |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2006/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `78.70% <0.00%> (+1.85%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2006?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/2006?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [beea315...ee42acd](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2006?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/config/eventing-kafka-broker/200-controller/100-config-kafka-leader-election.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-leader-election.yaml
index 8e1163e6a..910715958 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-leader-election.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-leader-election.yaml
@@ -45,3 +45,5 @@ data:
   leaseDuration: "15s"
   renewDeadline: "10s"
   retryPeriod: "2s"
+  map-lease-prefix.kafka-broker-controller.knative.dev.eventing-kafka-broker.control-plane.pkg.reconciler.source.Reconciler: kafka-controller.knative.dev.eventing-kafka.pkg.source.reconciler.source.reconciler
+  map-lease-prefix.kafka-broker-controller.knative.dev.eventing-kafka-broker.control-plane.pkg.reconciler.channel.Reconciler: kafkachannel-controller.knative.dev.eventing-kafka.pkg.channel.consolidated.reconciler.controller.reconciler
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
index 8887c0983..6f28419eb 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
@@ -40,6 +40,12 @@ rules:
       - update
       - get
       - watch
+  - apiGroups:
+      - "*"
+    resources:
+      - configmaps/finalizers
+    verbs:
+      - update
   - apiGroups:
       - "*"
     resources:
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/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index a863f291e..85275e243 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -67,8 +67,8 @@ Lists of 158 third-party dependencies.
      (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Server Mock (io.fabric8:kubernetes-server-mock:5.12.1 - http://fabric8.io/kubernetes-server-mock/)
      (Apache License, Version 2.0) Fabric8 :: Mock Web Server (io.fabric8:mockwebserver:0.2.2 - http://fabric8.io/)
      (The Apache Software License, Version 2.0) zjsonpatch (io.fabric8:zjsonpatch:0.3.0 - https://github.com/fabric8io/zjsonpatch/)
-     (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.8.3 - https://github.com/micrometer-metrics/micrometer)
-     (The Apache Software License, Version 2.0) micrometer-registry-prometheus (io.micrometer:micrometer-registry-prometheus:1.8.3 - https://github.com/micrometer-metrics/micrometer)
+     (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.8.4 - https://github.com/micrometer-metrics/micrometer)
+     (The Apache Software License, Version 2.0) micrometer-registry-prometheus (io.micrometer:micrometer-registry-prometheus:1.8.4 - https://github.com/micrometer-metrics/micrometer)
      (Apache License, Version 2.0) Netty/Buffer (io.netty:netty-buffer:4.1.74.Final - https://netty.io/netty-buffer/)
      (Apache License, Version 2.0) Netty/Codec (io.netty:netty-codec:4.1.74.Final - https://netty.io/netty-codec/)
      (Apache License, Version 2.0) Netty/Codec/DNS (io.netty:netty-codec-dns:4.1.74.Final - https://netty.io/netty-codec-dns/)
diff --git a/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml b/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml
index b146fba79..f691d02fb 100644
--- a/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml
+++ b/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml
@@ -64,7 +64,6 @@ data:
     # ssl.truststore.location=
     # ssl.truststore.password=
     allow.auto.create.topics=true
-    auto.offset.reset=earliest
     client.dns.lookup=use_all_dns_ips
     connections.max.idle.ms=540000
     default.api.timeout.ms=60000
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticle.java
index a56369274..b8f8be3d3 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticle.java
@@ -21,6 +21,7 @@ import io.vertx.core.CompositeFuture;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecords;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,10 +42,20 @@ public final class UnorderedConsumerVerticle extends BaseConsumerVerticle {
   // This shouldn't be more than 2000, which is the default max time allowed
   // to block a verticle thread.
   private static final Duration POLL_TIMEOUT = Duration.ofMillis(1000);
+  private final int maxPollRecords;
 
-  public UnorderedConsumerVerticle(Initializer initializer,
-                                   Set<String> topics) {
+  private int inFlightRecords;
+
+  public UnorderedConsumerVerticle(final Initializer initializer,
+                                   final Set<String> topics,
+                                   final int maxPollRecords) {
     super(initializer, topics);
+    if (maxPollRecords <= 0) {
+      this.maxPollRecords = 500;
+    } else {
+      this.maxPollRecords = maxPollRecords;
+    }
+    this.inFlightRecords = 0;
   }
 
   @Override
@@ -69,7 +80,17 @@ public final class UnorderedConsumerVerticle extends BaseConsumerVerticle {
    * with the consumer parameter `max.poll.records`, and it's critical to
    * control the memory consumption of the dispatcher.
    */
-  private void poll() {
+  private synchronized void poll() {
+    if (inFlightRecords >= maxPollRecords) {
+      logger.info(
+        "In flight records exceeds " + ConsumerConfig.MAX_POLL_RECORDS_CONFIG +
+          " waiting for response from subscriber before polling for new records {} {}",
+        keyValue(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords),
+        keyValue("records", inFlightRecords)
+      );
+      return;
+    }
+
     this.consumer
       .poll(POLL_TIMEOUT)
       .onSuccess(this::handleRecords)
@@ -81,11 +102,17 @@ public final class UnorderedConsumerVerticle extends BaseConsumerVerticle {
   }
 
   private void handleRecords(final KafkaConsumerRecords<Object, CloudEvent> records) {
-    final var futures = new ArrayList<Future>(records.size());
+    // We are not forcing the dispatcher to send less than `max.poll.records`
+    // requests because we don't want to keep records in-memory by waiting
+    // for responses.
     for (int i = 0; i < records.size(); i++) {
-      futures.add(this.recordDispatcher.dispatch(records.recordAt(i)));
+      this.inFlightRecords++;
+      this.recordDispatcher.dispatch(records.recordAt(i))
+        .onComplete(v -> {
+          this.inFlightRecords--;
+          poll();
+        });
     }
-    CompositeFuture.join(futures)
-      .onComplete(v -> poll());
+    poll();
   }
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
index 8b7503b8b..c974b4ddc 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
@@ -209,7 +209,12 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
      })
        .mapEmpty();
 
-    return getConsumerVerticle(deliveryOrder, initializer, new HashSet<>(resource.getTopicsList()));
+    return getConsumerVerticle(
+      deliveryOrder,
+      initializer,
+      new HashSet<>(resource.getTopicsList()),
+      consumerConfigs.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG)
+    );
   }
 
   private Filter getFilter(DataPlaneContract.Egress egress) {
@@ -348,10 +353,13 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
 
   private static AbstractVerticle getConsumerVerticle(final DeliveryOrder type,
                                                       final BaseConsumerVerticle.Initializer initializer,
-                                                      final Set<String> topics) {
+                                                      final Set<String> topics,
+                                                      final Object maxPollRecords) {
     return switch (type) {
       case ORDERED -> new OrderedConsumerVerticle(initializer, topics);
-      case UNORDERED -> new UnorderedConsumerVerticle(initializer, topics);
+      case UNORDERED -> new UnorderedConsumerVerticle(
+        initializer, topics, maxPollRecords == null ? 0 : Integer.parseInt(maxPollRecords.toString())
+      );
     };
   }
 
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index dc8778076..c20d054ab 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -43,7 +43,7 @@
     <!-- dependencies version -->
     <vertx.version>4.2.5</vertx.version>
     <cloudevents.sdk.version>2.3.0</cloudevents.sdk.version>
-    <micrometer.version>1.8.3</micrometer.version>
+    <micrometer.version>1.8.4</micrometer.version>
     <opentelemetry.version>1.12.0</opentelemetry.version>
     <jackson.version>2.13.2</jackson.version>
     <protobuf.version>3.19.4</protobuf.version>
diff --git a/hack/control-plane.sh b/hack/control-plane.sh
index ea5cac994..5d41b3c31 100644
--- a/hack/control-plane.sh
+++ b/hack/control-plane.sh
@@ -21,6 +21,6 @@ readonly CONTROL_PLANE_POST_INSTALL_CONFIG_DIR=control-plane/config/post-install
 
 # Note: do not change this function name, it's used during releases.
 function control_plane_setup() {
-  ko resolve ${KO_FLAGS} -Rf "${CONTROL_PLANE_CONFIG_DIR}" | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_CONTROL_PLANE_ARTIFACT}" || return $?
-  ko resolve ${KO_FLAGS} -Rf "${CONTROL_PLANE_POST_INSTALL_CONFIG_DIR}" | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}" || return $?
+  ko resolve ${KO_FLAGS} -Rf "${CONTROL_PLANE_CONFIG_DIR}" | "${LABEL_YAML_CMD[@]}" >"${EVENTING_KAFKA_CONTROL_PLANE_ARTIFACT}" || return $?
+  ko resolve ${KO_FLAGS} -Rf "${CONTROL_PLANE_POST_INSTALL_CONFIG_DIR}" | "${LABEL_YAML_CMD[@]}" >"${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}" || return $?
 }


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/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticleTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticleTest.java
index d94620c09..fa4143632 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticleTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticleTest.java
@@ -22,6 +22,6 @@ public class UnorderedConsumerVerticleTest extends AbstractConsumerVerticleTest
   @Override
   BaseConsumerVerticle createConsumerVerticle(
     BaseConsumerVerticle.Initializer initializer, Set<String> topics) {
-    return new UnorderedConsumerVerticle(initializer, topics);
+    return new UnorderedConsumerVerticle(initializer, topics, 5);
   }
 }
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index 6261e71e9..f883afbcb 100644
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -108,6 +108,7 @@ function build_components_from_source() {
   [ -f "${EVENTING_KAFKA_BROKER_ARTIFACT}" ] && rm "${EVENTING_KAFKA_BROKER_ARTIFACT}"
   [ -f "${EVENTING_KAFKA_SINK_ARTIFACT}" ] && rm "${EVENTING_KAFKA_SINK_ARTIFACT}"
   [ -f "${EVENTING_KAFKA_CHANNEL_ARTIFACT}" ] && rm "${EVENTING_KAFKA_CHANNEL_ARTIFACT}"
+  [ -f "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}" ] && rm "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}"
 
   header "Data plane setup"
   data_plane_setup || fail_test "Failed to set up data plane components"
diff --git a/test/e2e_new/features/leases.go b/test/e2e_new/features/leases.go
new file mode 100644
index 000000000..fc2d9b85a
--- /dev/null
+++ b/test/e2e_new/features/leases.go
@@ -0,0 +1,69 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package features
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/feature"
+)
+
+func KafkaChannelLease() *feature.Feature {
+	f := feature.NewFeatureNamed("KafkaChannel lease")
+
+	f.Assert("verify lease name", verifyLeaseAcquired(
+		"kafkachannel-controller.knative.dev.eventing-kafka.pkg.channel.consolidated.reconciler.controller.reconciler.00-of-01",
+	))
+
+	return f
+}
+
+func KafkaSourceLease() *feature.Feature {
+	f := feature.NewFeatureNamed("KafkaSource lease")
+
+	f.Assert("verify lease name", verifyLeaseAcquired(
+		"kafka-controller.knative.dev.eventing-kafka.pkg.source.reconciler.source.reconciler.00-of-01",
+	))
+
+	return f
+}
+
+func verifyLeaseAcquired(name string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		err := wait.Poll(time.Second, time.Minute, func() (done bool, err error) {
+			lease, err := kubeclient.Get(ctx).
+				CoordinationV1().
+				Leases(system.Namespace()).
+				Get(ctx, name, metav1.GetOptions{})
+			if err != nil {
+				t.Fatal("failed to get lease %s/%s: %v", system.Namespace(), name, err)
+			}
+
+			return lease.Spec.HolderIdentity != nil &&
+				strings.HasPrefix(*lease.Spec.HolderIdentity, "kafka-controller"), nil
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}
diff --git a/test/e2e_new/lease_test.go b/test/e2e_new/lease_test.go
new file mode 100644
index 000000000..9fb63c829
--- /dev/null
+++ b/test/e2e_new/lease_test.go
@@ -0,0 +1,46 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package e2e_new
+
+import (
+	"testing"
+
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+
+	"knative.dev/eventing-kafka-broker/test/e2e_new/features"
+)
+
+func TestLeaseAcquired(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, features.KafkaChannelLease())
+	env.Test(ctx, t, features.KafkaSourceLease())
+}