Repository: knative-sandbox/eventing-kafka-broker
Pull request #3177 (2023-06-27 19:18:53)

Migrate current dispatcher to use `ReactiveKafkaConsumer` instead of Vertx KafkaConsumer.
progress #2928 
The new dispatcher-vertx module is the one that uses Vertx implementation.

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Remove all the Vertx Kafka client dependencies from the base dispatcher module.
- Change the current `main()` to `start(..)` and call from the other module with specialised Consumer implementation as args.
- migrate all the tests from Direct KafkaConsumer to use ReactiveKafkaConsumer Interface.

## Aditional Info

- The implementation of passing handler function to `ConsumerRebalanceListener` has gotten complex. I will simplify that further in the future.
- Tests for the dispatcher-vertx module are not there as vertx implementations are already well tested.

knative-prow[bot]: Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with `/test all`
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/3177#" title="Author self-approved">debasishbsws</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign [pierdipi](https://github.com/pierdipi) for approval. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

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).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)**

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"]} -->
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/3177?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#3177](https://app.codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/3177?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (8ded997) into [main](https://app.codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/20a3799638c056f0ef7a77d86d4612e918c4fc68?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (20a3799) will **decrease** coverage by `4.95%`.
> The diff coverage is `n/a`.

```diff
@@             Coverage Diff              @@
##               main    #3177      +/-   ##
============================================
- Coverage     63.51%   58.57%   -4.95%     
============================================
  Files           167       90      -77     
  Lines         11660     8990    -2670     
  Branches        239        0     -239     
============================================
- Hits           7406     5266    -2140     
+ Misses         3701     3313     -388     
+ Partials        553      411     -142     
```

| 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.

[see 78 files with indirect coverage changes](https://app.codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/3177/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/3177#" title="Author self-approved">debasishbsws</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign [zroubalik](https://github.com/zroubalik) for approval. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

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).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)**

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":["zroubalik"]} -->
debasishbsws: @pierDipi In the last commit basically I create `ConsumerRebalanceListener` from the `PartisionsRevokeHandlers` and pass it to `ConsumerVirticleContext` and then while calling the `subscribe()` method I am passing that ConsumerRebalanceListener with it. There may by some redundancy now like in future I can change the PartisionsRevokeHandlers in order to reduce the complexity.
debasishbsws: Didn't look into the Benchmark module properly just migrate it to use Vertx implementation. 
debasishbsws: /test all
knative-prow[bot]: @debasishbsws: 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
--- | --- | --- | --- | ---
build-tests_eventing-kafka-broker_main | be13474aebf9fb977b45e833040c5129c4f49dd3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3177/build-tests_eventing-kafka-broker_main/1675116178143449088) | true | `/test build-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @debasishbsws: 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
--- | --- | --- | --- | ---
build-tests_eventing-kafka-broker_main | be13474aebf9fb977b45e833040c5129c4f49dd3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3177/build-tests_eventing-kafka-broker_main/1675116178143449088) | true | `/test build-tests`
channel-integration-tests-ssl_eventing-kafka-broker_main | be13474aebf9fb977b45e833040c5129c4f49dd3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3177/channel-integration-tests-ssl_eventing-kafka-broker_main/1675116178491576320) | true | `/test channel-integration-tests-ssl`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_main | be13474aebf9fb977b45e833040c5129c4f49dd3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3177/channel-integration-tests-ssl_eventing-kafka-broker_main/1675116178491576320) | true | `/test channel-integration-tests-ssl`
build-tests_eventing-kafka-broker_main | be13474aebf9fb977b45e833040c5129c4f49dd3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3177/build-tests_eventing-kafka-broker_main/1675468032903745536) | true | `/test build-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @debasishbsws: 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
--- | --- | --- | --- | ---
build-tests_eventing-kafka-broker_main | be13474aebf9fb977b45e833040c5129c4f49dd3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3177/build-tests_eventing-kafka-broker_main/1675468032903745536) | true | `/test build-tests`
channel-integration-tests-ssl_eventing-kafka-broker_main | be13474aebf9fb977b45e833040c5129c4f49dd3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3177/channel-integration-tests-ssl_eventing-kafka-broker_main/1675468032970854400) | true | `/test channel-integration-tests-ssl`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
debasishbsws: Though tests are saying to run `update-codegen.sh`, but after running that `DataPlaneContract.java`'s `hasReplyUrf()` method is being removed.
![image](https://github.com/knative-sandbox/eventing-kafka-broker/assets/65381620/52d23ca7-1f45-4615-ad4d-25a4f1d588c1)

But we need this as we call it from
https://github.com/knative-sandbox/eventing-kafka-broker/blob/fb89db60df3fdb7d6ae6721037a081b5d88a60c8/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleBuilder.java#L204-L206
knative-prow[bot]: @debasishbsws: 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
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_main | ca1dadee3bc77a685396f31ad1616825b4837717 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3177/channel-integration-tests-ssl_eventing-kafka-broker_main/1675479144533594112) | true | `/test channel-integration-tests-ssl`

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

<details>

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

Patch:
diff --git a/control-plane/pkg/prober/composite_prober.go b/control-plane/pkg/prober/composite_prober.go
new file mode 100644
index 000000000..7ebbafdec
--- /dev/null
+++ b/control-plane/pkg/prober/composite_prober.go
@@ -0,0 +1,67 @@
+/*
+ * Copyright 2023 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 prober
+
+import (
+	"context"
+	"net/http"
+)
+
+type compositeProber struct {
+	httpProber  Prober
+	httpsProber Prober
+}
+
+// NewComposite creates a composite prober.
+//
+// It reports status changes using the provided EnqueueFunc.
+func NewComposite(ctx context.Context, httpPort string, httpsPort string, IPsLister IPsLister, enqueue EnqueueFunc, caCerts *string) (NewProber, error) {
+	httpProber := NewAsync(ctx, http.DefaultClient, httpPort, IPsLister, enqueue)
+	httpsProber, err := NewAsyncWithTLS(ctx, httpsPort, IPsLister, enqueue, caCerts)
+	if err != nil {
+		return nil, err
+	}
+	return &compositeProber{
+		httpProber:  httpProber,
+		httpsProber: httpsProber,
+	}, nil
+}
+
+func (c *compositeProber) Probe(ctx context.Context, addressable NewAddressable, expected Status) Status {
+	var status Status
+	for _, addr := range addressable.AddressStatus.Addresses {
+		oldAddressable := Addressable{
+			ResourceKey: addressable.ResourceKey,
+			Address:     addr.URL.URL(),
+		}
+		if addr.URL.Scheme == "https" {
+			status = c.httpsProber.Probe(ctx, oldAddressable, expected)
+		} else if addr.URL.Scheme == "http" {
+			c.httpProber.Probe(ctx, oldAddressable, expected)
+		}
+		if status != expected {
+			return status
+		}
+	}
+	return expected
+}
+
+func (c *compositeProber) RotateRootCaCerts(caCerts *string) error {
+	// we don't need to rotate the certs on the http prober as it isn't using them
+	err := c.httpsProber.RotateRootCaCerts(caCerts)
+	return err
+}
diff --git a/control-plane/pkg/prober/prober.go b/control-plane/pkg/prober/prober.go
index 364f8e0cd..9855df397 100644
--- a/control-plane/pkg/prober/prober.go
+++ b/control-plane/pkg/prober/prober.go
@@ -25,6 +25,7 @@ import (
 
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/types"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/network"
 )
 
@@ -47,6 +48,22 @@ type Prober interface {
 	RotateRootCaCerts(caCerts *string) error
 }
 
+// NewAddressable contains addressable resource data for the new prober
+type NewAddressable struct {
+	// Addressable status
+	AddressStatus *duckv1.AddressStatus
+	// Resource key
+	ResourceKey types.NamespacedName
+}
+
+// NewProber probes an addressable resource
+type NewProber interface {
+	// Probe probes the provided NewAddressable resource and returns its Status
+	Probe(ctx context.Context, addressable NewAddressable, expected Status) Status
+	// RotateRootCaCerts rotates the CA certs used to make http requests
+	RotateRootCaCerts(caCerts *string) error
+}
+
 // Func type is an adapter to allow the use of
 // ordinary functions as Prober. If f is a function
 // with the appropriate signature, Func(f) is a
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 402cef5cb..5cf72e608 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -1,5 +1,5 @@
 
-Lists of 216 third-party dependencies.
+Lists of 217 third-party dependencies.
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Classic Module (ch.qos.logback:logback-classic:1.2.11 - http://logback.qos.ch/logback-classic)
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Core Module (ch.qos.logback:logback-core:1.2.11 - http://logback.qos.ch/logback-core)
      (Apache License 2.0) brotli4j (com.aayushatharva.brotli4j:brotli4j:1.8.0 - https://github.com/hyperxpro/Brotli4j/brotli4j)
@@ -33,6 +33,7 @@ Lists of 216 third-party dependencies.
      (Unknown license) contract (dev.knative.eventing.kafka.broker:contract:1.0-SNAPSHOT - no url defined)
      (Unknown license) core (dev.knative.eventing.kafka.broker:core:1.0-SNAPSHOT - no url defined)
      (Unknown license) dispatcher (dev.knative.eventing.kafka.broker:dispatcher:1.0-SNAPSHOT - no url defined)
+     (Unknown license) dispatcher-vertx (dev.knative.eventing.kafka.broker:dispatcher-vertx:1.0-SNAPSHOT - no url defined)
      (Unknown license) receiver (dev.knative.eventing.kafka.broker:receiver:1.0-SNAPSHOT - no url defined)
      (The Apache Software License, Version 2.0) CloudEvents - API (io.cloudevents:cloudevents-api:2.3.0 - https://cloudevents.github.io/sdk-java/cloudevents-api/)
      (The Apache Software License, Version 2.0) CloudEvents - Core (io.cloudevents:cloudevents-core:2.3.0 - https://cloudevents.github.io/sdk-java/cloudevents-core/)
diff --git a/data-plane/benchmarks/pom.xml b/data-plane/benchmarks/pom.xml
index 6837d817b..350d5f992 100644
--- a/data-plane/benchmarks/pom.xml
+++ b/data-plane/benchmarks/pom.xml
@@ -52,7 +52,7 @@
 
     <dependency>
       <groupId>dev.knative.eventing.kafka.broker</groupId>
-      <artifactId>dispatcher</artifactId>
+      <artifactId>dispatcher-vertx</artifactId>
       <version>${project.version}</version>
       <exclusions> <!-- Exclude logging and noop -->
         <exclusion>
diff --git a/data-plane/benchmarks/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManagerBenchmark.java b/data-plane/benchmarks/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManagerBenchmark.java
index e2f8b56a3..3f5374e32 100644
--- a/data-plane/benchmarks/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManagerBenchmark.java
+++ b/data-plane/benchmarks/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManagerBenchmark.java
@@ -16,20 +16,15 @@
 package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
 
 import io.cloudevents.CloudEvent;
-import io.vertx.core.AsyncResult;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
-import io.vertx.kafka.client.common.PartitionInfo;
-import io.vertx.kafka.client.common.TopicPartition;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecords;
-import io.vertx.kafka.client.consumer.KafkaReadStream;
-import io.vertx.kafka.client.consumer.OffsetAndMetadata;
-import io.vertx.kafka.client.consumer.OffsetAndTimestamp;
-import io.vertx.kafka.client.consumer.impl.KafkaConsumerRecordImpl;
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.Level;
 import org.openjdk.jmh.annotations.Scope;
@@ -37,34 +32,32 @@ import org.openjdk.jmh.annotations.Setup;
 import org.openjdk.jmh.annotations.State;
 import org.openjdk.jmh.infra.Blackhole;
 
+import dev.knative.eventing.kafka.broker.dispatcher.ReactiveKafkaConsumer;
+
 import java.time.Duration;
-import java.util.List;
+import java.util.Collection;
 import java.util.Map;
-import java.util.Set;
-import java.util.regex.Pattern;
 
 public class UnorderedOffsetManagerBenchmark {
 
   @State(Scope.Thread)
   public static class RecordsState {
 
-    private KafkaConsumerRecord<String, CloudEvent>[][] records;
+    private ConsumerRecord<String, CloudEvent>[][] records;
 
     @Setup(Level.Trial)
     @SuppressWarnings("unchecked")
     public void doSetup() {
-      this.records = new KafkaConsumerRecord[100][10_000];
+      this.records = new ConsumerRecord[100][10_000];
       for (int p = 0; p < 100; p++) {
         for (int o = 0; o < 10_000; o++) {
-          this.records[p][o] = new KafkaConsumerRecordImpl<>(
-            new ConsumerRecord<>(
+          this.records[p][o] = new ConsumerRecord<>(
               "abc",
               p,
               o,
               null,
               null
-            )
-          );
+            );
         }
       }
     }
@@ -140,336 +133,13 @@ public class UnorderedOffsetManagerBenchmark {
     }
   }
 
-  static class MockKafkaConsumer implements io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> {
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> exceptionHandler(
-      Handler<Throwable> handler) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> handler(
-      Handler<KafkaConsumerRecord<String, CloudEvent>> handler) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> pause() {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> resume() {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> fetch(long amount) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> endHandler(
-      Handler<Void> endHandler) {
-      return null;
-    }
-
-    @Override
-    public long demand() {
-      return 0;
-    }
-
-    @Override
-    public Future<Void> subscribe(String topic) {
-      return null;
-    }
-
-    @Override
-    public Future<Void> subscribe(Set<String> topics) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> subscribe(
-      String topic,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> subscribe(
-      Set<String> topics,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public Future<Void> subscribe(Pattern pattern) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> subscribe(
-      Pattern pattern,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public Future<Void> assign(TopicPartition topicPartition) {
-      return null;
-    }
-
-    @Override
-    public Future<Void> assign(
-      Set<TopicPartition> topicPartitions) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> assign(
-      TopicPartition topicPartition,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> assign(
-      Set<TopicPartition> topicPartitions,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> assignment(
-      Handler<AsyncResult<Set<TopicPartition>>> handler) {
-      return null;
-    }
-
-    @Override
-    public Future<Set<TopicPartition>> assignment() {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> listTopics(
-      Handler<AsyncResult<Map<String, List<PartitionInfo>>>> handler) {
-      return null;
-    }
-
-    @Override
-    public Future<Map<String, List<PartitionInfo>>> listTopics() {
-      return null;
-    }
-
-    @Override
-    public Future<Void> unsubscribe() {
-      return null;
-    }
 
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> unsubscribe(
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> subscription(
-      Handler<AsyncResult<Set<String>>> handler) {
-      return null;
-    }
-
-    @Override
-    public Future<Set<String>> subscription() {
-      return null;
-    }
-
-    @Override
-    public Future<Void> pause(TopicPartition topicPartition) {
-      return null;
-    }
-
-    @Override
-    public Future<Void> pause(
-      Set<TopicPartition> topicPartitions) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> pause(
-      TopicPartition topicPartition,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> pause(
-      Set<TopicPartition> topicPartitions,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public void paused(
-      Handler<AsyncResult<Set<TopicPartition>>> handler) {
-
-    }
-
-    @Override
-    public Future<Set<TopicPartition>> paused() {
-      return null;
-    }
-
-    @Override
-    public Future<Void> resume(TopicPartition topicPartition) {
-      return null;
-    }
-
-    @Override
-    public Future<Void> resume(
-      Set<TopicPartition> topicPartitions) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> resume(
-      TopicPartition topicPartition,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> resume(
-      Set<TopicPartition> topicPartitions,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> partitionsRevokedHandler(
-      Handler<Set<TopicPartition>> handler) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> partitionsAssignedHandler(
-      Handler<Set<TopicPartition>> handler) {
-      return null;
-    }
-
-    @Override
-    public Future<Void> seek(TopicPartition topicPartition, long offset) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> seek(
-      TopicPartition topicPartition, long offset,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public Future<Void> seekToBeginning(TopicPartition topicPartition) {
-      return null;
-    }
-
-    @Override
-    public Future<Void> seekToBeginning(
-      Set<TopicPartition> topicPartitions) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> seekToBeginning(
-      TopicPartition topicPartition,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> seekToBeginning(
-      Set<TopicPartition> topicPartitions,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public Future<Void> seekToEnd(TopicPartition topicPartition) {
-      return null;
-    }
-
-    @Override
-    public Future<Void> seekToEnd(
-      Set<TopicPartition> topicPartitions) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> seekToEnd(
-      TopicPartition topicPartition,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> seekToEnd(
-      Set<TopicPartition> topicPartitions,
-      Handler<AsyncResult<Void>> completionHandler) {
-      return null;
-    }
-
-    @Override
-    public Future<Void> commit() {
-      return null;
-    }
-
-    @Override
-    public void commit(
-      Handler<AsyncResult<Void>> completionHandler) {
-
-    }
+  static class MockKafkaConsumer implements ReactiveKafkaConsumer<String, CloudEvent> {
 
     @Override
     public Future<Map<TopicPartition, OffsetAndMetadata>> commit(
-      Map<TopicPartition, OffsetAndMetadata> offsets) {
-      return Future.succeededFuture(offsets);
-    }
-
-    @Override
-    public void commit(
-      Map<TopicPartition, OffsetAndMetadata> offsets,
-      Handler<AsyncResult<Map<TopicPartition, OffsetAndMetadata>>> completionHandler) {
-      completionHandler.handle(Future.succeededFuture(offsets));
-    }
-
-    @Override
-    public void committed(TopicPartition topicPartition,
-                          Handler<AsyncResult<OffsetAndMetadata>> handler) {
+        Map<TopicPartition, OffsetAndMetadata> offset) {
 
-    }
-
-    @Override
-    public Future<OffsetAndMetadata> committed(
-      TopicPartition topicPartition) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> partitionsFor(
-      String topic,
-      Handler<AsyncResult<List<PartitionInfo>>> handler) {
-      return null;
-    }
-
-    @Override
-    public Future<List<PartitionInfo>> partitionsFor(String topic) {
-      return null;
-    }
-
-    @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> batchHandler(
-      Handler<KafkaConsumerRecords<String, CloudEvent>> handler) {
       return null;
     }
 
@@ -479,96 +149,27 @@ public class UnorderedOffsetManagerBenchmark {
     }
 
     @Override
-    public void close(
-      Handler<AsyncResult<Void>> completionHandler) {
-
-    }
-
-    @Override
-    public void position(TopicPartition partition,
-                         Handler<AsyncResult<Long>> handler) {
-
-    }
-
-    @Override
-    public Future<Long> position(TopicPartition partition) {
-      return null;
-    }
-
-    @Override
-    public void offsetsForTimes(
-      Map<TopicPartition, Long> topicPartitionTimestamps,
-      Handler<AsyncResult<Map<TopicPartition, OffsetAndTimestamp>>> handler) {
-
-    }
-
-    @Override
-    public Future<Map<TopicPartition, OffsetAndTimestamp>> offsetsForTimes(
-      Map<TopicPartition, Long> topicPartitionTimestamps) {
-      return null;
-    }
-
-    @Override
-    public void offsetsForTimes(TopicPartition topicPartition, Long timestamp,
-                                Handler<AsyncResult<OffsetAndTimestamp>> handler) {
-
-    }
-
-    @Override
-    public Future<OffsetAndTimestamp> offsetsForTimes(
-      TopicPartition topicPartition, Long timestamp) {
+    public Future<Void> pause(Collection<TopicPartition> partitions) {
       return null;
     }
 
     @Override
-    public void beginningOffsets(
-      Set<TopicPartition> topicPartitions,
-      Handler<AsyncResult<Map<TopicPartition, Long>>> handler) {
-
-    }
-
-    @Override
-    public Future<Map<TopicPartition, Long>> beginningOffsets(
-      Set<TopicPartition> topicPartitions) {
+    public Future<ConsumerRecords<String, CloudEvent>> poll(Duration timeout) {
       return null;
     }
 
     @Override
-    public void beginningOffsets(TopicPartition topicPartition,
-                                 Handler<AsyncResult<Long>> handler) {
-
-    }
-
-    @Override
-    public Future<Long> beginningOffsets(TopicPartition topicPartition) {
+    public Future<Void> resume(Collection<TopicPartition> partitions) {
       return null;
     }
 
     @Override
-    public void endOffsets(Set<TopicPartition> topicPartitions,
-                           Handler<AsyncResult<Map<TopicPartition, Long>>> handler) {
-
-    }
-
-    @Override
-    public Future<Map<TopicPartition, Long>> endOffsets(
-      Set<TopicPartition> topicPartitions) {
+    public Future<Void> subscribe(Collection<String> topics) {
       return null;
     }
 
     @Override
-    public void endOffsets(TopicPartition topicPartition,
-                           Handler<AsyncResult<Long>> handler) {
-
-    }
-
-    @Override
-    public Future<Long> endOffsets(TopicPartition topicPartition) {
-      return null;
-    }
-
-    @Override
-    public KafkaReadStream<String, CloudEvent> asStream() {
+    public Future<Void> subscribe(Collection<String> topics, ConsumerRebalanceListener listener) {
       return null;
     }
 
@@ -578,21 +179,9 @@ public class UnorderedOffsetManagerBenchmark {
     }
 
     @Override
-    public io.vertx.kafka.client.consumer.KafkaConsumer<String, CloudEvent> pollTimeout(
-      Duration timeout) {
+    public ReactiveKafkaConsumer<String, CloudEvent> exceptionHandler(Handler<Throwable> handler) {
       return null;
     }
 
-    @Override
-    public void poll(Duration timeout,
-                     Handler<AsyncResult<KafkaConsumerRecords<String, CloudEvent>>> handler) {
-
-    }
-
-    @Override
-    public Future<KafkaConsumerRecords<String, CloudEvent>> poll(Duration timeout) {
-      return null;
-    }
   }
-
 }
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutor.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutor.java
index c8e060bb7..8b3fb8ad3 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutor.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutor.java
@@ -22,7 +22,7 @@ import io.micrometer.core.instrument.Gauge;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Tags;
 import io.vertx.core.Future;
-import io.vertx.kafka.client.common.TopicPartition;
+import org.apache.kafka.common.TopicPartition;
 
 import java.util.ArrayDeque;
 import java.util.Queue;
@@ -56,8 +56,8 @@ public class OrderedAsyncExecutor {
 
     if (meterRegistry != null && egress != null && egress.getFeatureFlags().getEnableOrderedExecutorMetrics()) {
       final var tags = Tags.of(
-        Metrics.Tags.PARTITION_ID, topicPartition.getPartition() + "",
-        Metrics.Tags.TOPIC_ID, topicPartition.getTopic(),
+        Metrics.Tags.PARTITION_ID, topicPartition.partition() + "",
+        Metrics.Tags.TOPIC_ID, topicPartition.topic(),
         Metrics.Tags.CONSUMER_NAME, egress.getReference().getName(),
         Metrics.Tags.RESOURCE_NAMESPACE, egress.getReference().getNamespace()
       );
diff --git a/data-plane/dispatcher-vertx/pom.xml b/data-plane/dispatcher-vertx/pom.xml
index 265aac154..77913e73f 100644
--- a/data-plane/dispatcher-vertx/pom.xml
+++ b/data-plane/dispatcher-vertx/pom.xml
@@ -36,10 +36,63 @@
     </dependency>
   </dependencies>
 
-  <!-- <build>
+  <build>
     <plugins>
-
+      <plugin>
+        <groupId>com.google.cloud.tools</groupId>
+        <artifactId>jib-maven-plugin</artifactId>
+        <version>${jib.version}</version>
+        <configuration>
+          <to>
+            <image>${env.KO_DOCKER_REPO}/knative-kafka-broker-dispatcher:${env.TAG}</image>
+          </to>
+        </configuration>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-shade-plugin</artifactId>
+        <version>${maven.shade.plugin.version}</version>
+        <configuration>
+          <minimizeJar>true</minimizeJar>
+        </configuration>
+        <executions>
+          <execution>
+            <phase>package</phase>
+            <goals>
+              <goal>shade</goal>
+            </goals>
+            <configuration>
+              <transformers>
+                <!-- This merges all the META-INF/services -->
+                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+                <transformer
+                  implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+                  <mainClass>dev.knative.eventing.kafka.broker.dispatchervertx.Main</mainClass>
+                </transformer>
+              </transformers>
+              <filters>
+                <filter>
+                  <artifact>*:*</artifact>
+                  <excludeDefaults>false</excludeDefaults>
+                </filter>
+                <filter>
+                  <artifact>net.logstash.logback:logstash-logback-encoder</artifact>
+                  <includes>
+                    <include>**</include>
+                  </includes>
+                </filter>
+                <filter>
+                  <artifact>org.apache.kafka:kafka-clients</artifact>
+                  <includes>
+                    <include>**</include>
+                  </includes>
+                </filter>
+              </filters>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
     </plugins>
-  </build> -->
+  </build>
 
 </project>
diff --git a/data-plane/dispatcher-vertx/src/main/java/dev/knative/eventing/kafka/broker/dispatchervertx/Main.java b/data-plane/dispatcher-vertx/src/main/java/dev/knative/eventing/kafka/broker/dispatchervertx/Main.java
new file mode 100644
index 000000000..fb0041212
--- /dev/null
+++ b/data-plane/dispatcher-vertx/src/main/java/dev/knative/eventing/kafka/broker/dispatchervertx/Main.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * 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 dev.knative.eventing.kafka.broker.dispatchervertx;
+
+import java.io.IOException;
+
+public class Main {
+    public static void main(String[] args) throws IOException {
+        dev.knative.eventing.kafka.broker.dispatcher.main.Main.start(args, new VertxConsumerFactory<>());
+    }
+}
diff --git a/data-plane/dispatcher-vertx/src/main/java/dev/knative/eventing/kafka/broker/dispatchervertx/VertxKafkaConsumer.java b/data-plane/dispatcher-vertx/src/main/java/dev/knative/eventing/kafka/broker/dispatchervertx/VertxKafkaConsumer.java
index 237f20095..fda234cb2 100644
--- a/data-plane/dispatcher-vertx/src/main/java/dev/knative/eventing/kafka/broker/dispatchervertx/VertxKafkaConsumer.java
+++ b/data-plane/dispatcher-vertx/src/main/java/dev/knative/eventing/kafka/broker/dispatchervertx/VertxKafkaConsumer.java
@@ -19,9 +19,11 @@ import java.time.Duration;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
@@ -90,6 +92,21 @@ public class VertxKafkaConsumer<K, V> implements ReactiveKafkaConsumer<K, V> {
         return consumer.subscribe(new HashSet<>(topics));
     }
 
+    @Override
+    public Future<Void> subscribe(Collection<String> topics, ConsumerRebalanceListener listener) {
+        Handler<Set<io.vertx.kafka.client.common.TopicPartition>> handler = partitions -> {
+            Set<TopicPartition> apachePartitions = new HashSet<>();
+            for (io.vertx.kafka.client.common.TopicPartition vertxPartition : partitions) {
+                apachePartitions.add(new TopicPartition(vertxPartition.getTopic(), vertxPartition.getPartition()));
+            }
+
+            listener.onPartitionsRevoked(apachePartitions);
+        };
+        consumer = consumer.partitionsRevokedHandler(handler);
+
+        return consumer.subscribe(new HashSet<>(topics));
+    }
+
     @Override
     public Consumer<K, V> unwrap() {
         return consumer.unwrap();
diff --git a/data-plane/dispatcher/pom.xml b/data-plane/dispatcher/pom.xml
index 63da39c2a..9a1f5ed28 100644
--- a/data-plane/dispatcher/pom.xml
+++ b/data-plane/dispatcher/pom.xml
@@ -82,57 +82,14 @@
 
   <build>
     <plugins>
-      <plugin>
-        <groupId>com.google.cloud.tools</groupId>
-        <artifactId>jib-maven-plugin</artifactId>
-        <version>${jib.version}</version>
-        <configuration>
-          <to>
-            <image>${env.KO_DOCKER_REPO}/knative-kafka-broker-dispatcher:${env.TAG}</image>
-          </to>
-        </configuration>
-      </plugin>
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
-        <artifactId>maven-shade-plugin</artifactId>
-        <version>${maven.shade.plugin.version}</version>
-        <configuration>
-          <minimizeJar>true</minimizeJar>
-        </configuration>
+        <artifactId>maven-jar-plugin</artifactId>
         <executions>
           <execution>
-            <phase>package</phase>
             <goals>
-              <goal>shade</goal>
+              <goal>test-jar</goal>
             </goals>
-            <configuration>
-              <transformers>
-                <!-- This merges all the META-INF/services -->
-                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
-                <transformer
-                  implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
-                  <mainClass>dev.knative.eventing.kafka.broker.dispatcher.main.Main</mainClass>
-                </transformer>
-              </transformers>
-              <filters>
-                <filter>
-                  <artifact>*:*</artifact>
-                  <excludeDefaults>false</excludeDefaults>
-                </filter>
-                <filter>
-                  <artifact>net.logstash.logback:logstash-logback-encoder</artifact>
-                  <includes>
-                    <include>**</include>
-                  </includes>
-                </filter>
-                <filter>
-                  <artifact>org.apache.kafka:kafka-clients</artifact>
-                  <includes>
-                    <include>**</include>
-                  </includes>
-                </filter>
-              </filters>
-            </configuration>
           </execution>
         </executions>
       </plugin>
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/ReactiveKafkaConsumer.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/ReactiveKafkaConsumer.java
index 66c8bf287..737e0e79b 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/ReactiveKafkaConsumer.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/ReactiveKafkaConsumer.java
@@ -20,6 +20,7 @@ import java.util.Collection;
 import java.util.Map;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
@@ -81,6 +82,15 @@ public interface ReactiveKafkaConsumer<K, V> {
      */
     Future<Void> subscribe(Collection<String> topics);
 
+    /**
+     * Subscribes to the specified topics to start consuming from them.
+     *
+     * @param topics   The topics to subscribe to.
+     * @param listener The listener to invoke when the consumer is rebalanced.
+     * @return A future indicating the success or failure of the subscribe operation.
+     */
+    Future<Void> subscribe(Collection<String> topics, ConsumerRebalanceListener listener);
+
     /**
      * Retrieves the underlying Kafka Consumer instance.
      *
@@ -95,4 +105,5 @@ public interface ReactiveKafkaConsumer<K, V> {
      * @return This consumer instance with the exception handler set.
      */
     ReactiveKafkaConsumer<K, V> exceptionHandler(Handler<Throwable> handler);
+
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcher.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcher.java
index b264922e3..54de07077 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcher.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcher.java
@@ -15,10 +15,12 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher;
 
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
 import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
 import io.cloudevents.CloudEvent;
 import io.vertx.core.Future;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
+
 
 /**
  * This interface performs the dispatch of consumed records.
@@ -32,6 +34,6 @@ public interface RecordDispatcher extends AsyncCloseable {
    * @param record record to handle.
    * @return the completion future.
    */
-  Future<Void> dispatch(KafkaConsumerRecord<Object, CloudEvent> record);
+  Future<Void> dispatch(ConsumerRecord<Object, CloudEvent> record);
 
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java
index 7898c15ff..d85d9ab24 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java
@@ -15,8 +15,9 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher;
 
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
 import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 
 /**
  * This class contains hooks for listening events through the {@link dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher} lifecycle.
@@ -28,7 +29,7 @@ public interface RecordDispatcherListener extends AsyncCloseable {
    *
    * @param record record received.
    */
-  void recordReceived(KafkaConsumerRecord<?, ?> record);
+  void recordReceived(ConsumerRecord<?, ?> record);
 
   /**
    * The given record cannot be delivered to dead letter sink.
@@ -36,26 +37,26 @@ public interface RecordDispatcherListener extends AsyncCloseable {
    * @param record record undeliverable to dead letter sink.
    * @param ex     exception occurred.
    */
-  void failedToSendToDeadLetterSink(KafkaConsumerRecord<?, ?> record, Throwable ex);
+  void failedToSendToDeadLetterSink(ConsumerRecord<?, ?> record, Throwable ex);
 
   /**
    * The given event doesn't pass the filter.
    *
    * @param record record discarded.
    */
-  void recordDiscarded(KafkaConsumerRecord<?, ?> record);
+  void recordDiscarded(ConsumerRecord<?, ?> record);
 
   /**
    * The given record has been successfully sent to subscriber.
    *
    * @param record record sent to subscriber.
    */
-  void successfullySentToSubscriber(KafkaConsumerRecord<?, ?> record);
+  void successfullySentToSubscriber(ConsumerRecord<?, ?> record);
 
   /**
    * The given record has been successfully sent to dead letter sink.
    *
    * @param record record sent to dead letter sink.
    */
-  void successfullySentToDeadLetterSink(KafkaConsumerRecord<?, ?> record);
+  void successfullySentToDeadLetterSink(ConsumerRecord<?, ?> record);
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ConsumerRecordContext.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ConsumerRecordContext.java
index 74bc14f66..026bca391 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ConsumerRecordContext.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/ConsumerRecordContext.java
@@ -15,15 +15,16 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.impl;
 
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import io.cloudevents.CloudEvent;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
+
 
 class ConsumerRecordContext {
 
-  private KafkaConsumerRecord<Object, CloudEvent> record;
+  private ConsumerRecord<Object, CloudEvent> record;
   private long receivedAtMs;
 
-  ConsumerRecordContext(KafkaConsumerRecord<Object, CloudEvent> record) {
+  ConsumerRecordContext(ConsumerRecord<Object, CloudEvent> record) {
     this.record = record;
     this.resetTimer();
   }
@@ -36,11 +37,11 @@ class ConsumerRecordContext {
     return System.currentTimeMillis() - receivedAtMs;
   }
 
-  KafkaConsumerRecord<Object, CloudEvent> getRecord() {
+  ConsumerRecord<Object, CloudEvent> getRecord() {
     return record;
   }
 
-  void setRecord(final KafkaConsumerRecord<Object, CloudEvent> record) {
+  void setRecord(final ConsumerRecord<Object, CloudEvent> record) {
     this.record = record;
   }
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
index e274cbe67..d93e4e18e 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
@@ -38,7 +38,7 @@ import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.ext.web.client.HttpResponse;
 import io.vertx.kafka.client.common.tracing.ConsumerTracer;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
+// import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 import io.vertx.kafka.client.consumer.impl.KafkaConsumerRecordImpl;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.slf4j.Logger;
@@ -82,8 +82,8 @@ public class RecordDispatcherImpl implements RecordDispatcher {
   private static final int KN_ERROR_DATA_MAX_BYTES = 1024;
 
   private final Filter filter;
-  private final Function<KafkaConsumerRecord<Object, CloudEvent>, Future<HttpResponse<?>>> subscriberSender;
-  private final Function<KafkaConsumerRecord<Object, CloudEvent>, Future<HttpResponse<?>>> dlsSender;
+  private final Function<ConsumerRecord<Object, CloudEvent>, Future<HttpResponse<?>>> subscriberSender;
+  private final Function<ConsumerRecord<Object, CloudEvent>, Future<HttpResponse<?>>> dlsSender;
   private final RecordDispatcherListener recordDispatcherListener;
   private final AsyncCloseable closeable;
   private final ConsumerTracer consumerTracer;
@@ -140,7 +140,7 @@ public class RecordDispatcherImpl implements RecordDispatcher {
    * @param record record to handle.
    */
   @Override
-  public Future<Void> dispatch(KafkaConsumerRecord<Object, CloudEvent> record) {
+  public Future<Void> dispatch(ConsumerRecord<Object, CloudEvent> record) {
     if (closed.get()) {
       return Future.failedFuture("Dispatcher closed");
     }
@@ -168,7 +168,7 @@ public class RecordDispatcherImpl implements RecordDispatcher {
      */
     final var recordContext = new ConsumerRecordContext(record);
 
-    if (record.record().value() instanceof InvalidCloudEvent) {
+    if (record.value() instanceof InvalidCloudEvent) {
       incrementDiscardedRecord(recordContext);
       final var msg = String.format("Invalid record received topic %s, partition %d, offset %d", record.topic(), record.partition(), record.offset());
       logger.error(msg);
@@ -323,22 +323,20 @@ public class RecordDispatcherImpl implements RecordDispatcher {
     final var transformedCloudEvent = builder.build();
 
     final var cr = new ConsumerRecord<>(
-      recordContext.getRecord().record().topic(),
-      recordContext.getRecord().record().partition(),
-      recordContext.getRecord().record().offset(),
-      recordContext.getRecord().record().timestamp(),
-      recordContext.getRecord().record().timestampType(),
-      recordContext.getRecord().record().serializedKeySize(),
-      recordContext.getRecord().record().serializedValueSize(),
-      recordContext.getRecord().record().key(),
+      recordContext.getRecord().topic(),
+      recordContext.getRecord().partition(),
+      recordContext.getRecord().offset(),
+      recordContext.getRecord().timestamp(),
+      recordContext.getRecord().timestampType(),
+      recordContext.getRecord().serializedKeySize(),
+      recordContext.getRecord().serializedValueSize(),
+      recordContext.getRecord().key(),
       transformedCloudEvent,
-      recordContext.getRecord().record().headers(),
-      recordContext.getRecord().record().leaderEpoch()
+      recordContext.getRecord().headers(),
+      recordContext.getRecord().leaderEpoch()
     );
 
-    final var kcr = new KafkaConsumerRecordImpl<>(cr);
-
-    return new ConsumerRecordContext(kcr);
+    return new ConsumerRecordContext(cr);
   }
 
   private void onDeadLetterSinkSuccess(final ConsumerRecordContext recordContext,
@@ -369,12 +367,12 @@ public class RecordDispatcherImpl implements RecordDispatcher {
     // That means that we get a record with a null value and some CE
     // headers even though the record is a valid CloudEvent.
     logDebug("Value is null", recordContext.getRecord());
-    final var value = cloudEventDeserializer.deserialize(recordContext.getRecord().record().topic(), recordContext.getRecord().record().headers(), null);
-    recordContext.setRecord(new KafkaConsumerRecordImpl<>(KafkaConsumerRecordUtils.copyRecordAssigningValue(recordContext.getRecord().record(), value)));
+    final var value = cloudEventDeserializer.deserialize(recordContext.getRecord().topic(), recordContext.getRecord().headers(), null);
+    recordContext.setRecord(KafkaConsumerRecordUtils.copyRecordAssigningValue(recordContext.getRecord(), value));
     return recordContext;
   }
 
-  private Function<KafkaConsumerRecord<Object, CloudEvent>, Future<HttpResponse<?>>> composeSenderAndSinkHandler(
+  private Function<ConsumerRecord<Object, CloudEvent>, Future<HttpResponse<?>>> composeSenderAndSinkHandler(
     CloudEventSender sender, ResponseHandler sinkHandler, String senderType) {
     return rec -> sender.send(rec.value())
       .onFailure(ex -> logError("Failed to send event to " + senderType, rec, ex))
@@ -446,7 +444,7 @@ public class RecordDispatcherImpl implements RecordDispatcher {
   }
 
   private Tags getTags(final ConsumerRecordContext recordContext) {
-    if (recordContext.getRecord().record().value() instanceof InvalidCloudEvent) {
+    if (recordContext.getRecord().value() instanceof InvalidCloudEvent) {
       return this.consumerVerticleContext.getTags().and(INVALID_EVENT_TYPE_TAG);
     }
     return this.consumerVerticleContext.getTags().and(
@@ -456,7 +454,7 @@ public class RecordDispatcherImpl implements RecordDispatcher {
 
   private void logError(
     final String msg,
-    final KafkaConsumerRecord<Object, CloudEvent> record,
+    final ConsumerRecord<Object, CloudEvent> record,
     final Throwable cause) {
 
     if (logger.isDebugEnabled()) {
@@ -482,7 +480,7 @@ public class RecordDispatcherImpl implements RecordDispatcher {
 
   private void logDebug(
     final String msg,
-    final KafkaConsumerRecord<Object, CloudEvent> record) {
+    final ConsumerRecord<Object, CloudEvent> record) {
 
     logger.debug(msg + " {} {} {} {} {} {} {}",
       consumerVerticleContext.getLoggingKeyValue(),
@@ -497,7 +495,7 @@ public class RecordDispatcherImpl implements RecordDispatcher {
 
   private ConsumerTracer.StartedSpan getStartedSpan(ConsumerRecordContext recordContext, Context context) {
     if (consumerTracer != null) {
-      return consumerTracer.prepareMessageReceived(context, recordContext.getRecord().record());
+      return consumerTracer.prepareMessageReceived(context, recordContext.getRecord());
     }
     return null;
   }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChain.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChain.java
index 31687c53f..36cb15eda 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChain.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChain.java
@@ -19,8 +19,6 @@ import dev.knative.eventing.kafka.broker.dispatcher.CloudEventMutator;
 import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher;
 import io.cloudevents.CloudEvent;
 import io.vertx.core.Future;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
-import io.vertx.kafka.client.consumer.impl.KafkaConsumerRecordImpl;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 /**
@@ -39,9 +37,8 @@ public class RecordDispatcherMutatorChain implements RecordDispatcher {
   }
 
   @Override
-  public Future<Void> dispatch(KafkaConsumerRecord<Object, CloudEvent> kafkaRecord) {
-    final var record = kafkaRecord.record();
-    final var newRecord = new ConsumerRecord<>(
+  public Future<Void> dispatch(ConsumerRecord<Object, CloudEvent> record) {
+      final var newRecord = new ConsumerRecord<>(
       record.topic(),
       record.partition(),
       record.offset(),
@@ -54,7 +51,7 @@ public class RecordDispatcherMutatorChain implements RecordDispatcher {
       record.headers(),
       record.leaderEpoch()
     );
-    return next.dispatch(new KafkaConsumerRecordImpl<>(newRecord));
+    return next.dispatch(newRecord);
   }
 
   @Override
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/ConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/ConsumerVerticle.java
index a483d45cd..d52dd2533 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/ConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/ConsumerVerticle.java
@@ -16,6 +16,7 @@
 package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
 
 import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
+import dev.knative.eventing.kafka.broker.dispatcher.ReactiveKafkaConsumer;
 import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher;
 import dev.knative.eventing.kafka.broker.dispatcher.main.ConsumerVerticleContext;
 import io.cloudevents.CloudEvent;
@@ -23,7 +24,6 @@ import io.vertx.core.AbstractVerticle;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
-import io.vertx.kafka.client.consumer.KafkaConsumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -42,7 +42,7 @@ public abstract class ConsumerVerticle extends AbstractVerticle {
 
   private final ConsumerVerticleContext consumerVerticleContext;
 
-  KafkaConsumer<Object, CloudEvent> consumer;
+  ReactiveKafkaConsumer<Object, CloudEvent> consumer;
   RecordDispatcher recordDispatcher;
   private AsyncCloseable closeable;
 
@@ -79,7 +79,7 @@ public abstract class ConsumerVerticle extends AbstractVerticle {
     stopPromise.tryComplete();
   }
 
-  public void setConsumer(KafkaConsumer<Object, CloudEvent> consumer) {
+  public void setConsumer(ReactiveKafkaConsumer<Object, CloudEvent> consumer) {
     this.consumer = consumer;
   }
 
@@ -104,5 +104,5 @@ public abstract class ConsumerVerticle extends AbstractVerticle {
     return consumerVerticleContext;
   }
 
-  public abstract PartitionRevokedHandler getPartitionsRevokedHandler();
+  public abstract PartitionRevokedHandler getPartitionRevokedHandler();
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
index 28e5200ad..5923099e0 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
@@ -15,23 +15,22 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
 
+import dev.knative.eventing.kafka.broker.dispatcher.ReactiveKafkaConsumer;
 import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import io.vertx.core.CompositeFuture;
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
-import io.vertx.kafka.client.common.TopicPartition;
-import io.vertx.kafka.client.consumer.KafkaConsumer;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
-import io.vertx.kafka.client.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Arrays;
 import java.util.BitSet;
-import java.util.HashMap;
+import java.util.Collection;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
@@ -45,7 +44,7 @@ public final class OffsetManager implements RecordDispatcherListener {
 
   private static final Logger logger = LoggerFactory.getLogger(OffsetManager.class);
 
-  private final KafkaConsumer<?, ?> consumer;
+  private final ReactiveKafkaConsumer<?, ?> consumer;
 
   private final Map<TopicPartition, OffsetTracker> offsetTrackers;
 
@@ -61,7 +60,7 @@ public final class OffsetManager implements RecordDispatcherListener {
    * @param onCommit Callback invoked when an offset is actually committed
    */
   public OffsetManager(final Vertx vertx,
-                       final KafkaConsumer<?, ?> consumer,
+                       final ReactiveKafkaConsumer<?, ?> consumer,
                        final Consumer<Integer> onCommit,
                        final long commitIntervalMs) {
     Objects.requireNonNull(consumer, "provide consumer");
@@ -101,7 +100,7 @@ public final class OffsetManager implements RecordDispatcherListener {
    * @return
    */
   @Override
-  public void recordReceived(final KafkaConsumerRecord<?, ?> record) {
+  public void recordReceived(final ConsumerRecord<?, ?> record) {
     final var tp = new TopicPartition(record.topic(), record.partition());
     if (!offsetTrackers.containsKey(tp)) {
       // Initialize offset tracker for the given record's topic/partition.
@@ -113,7 +112,7 @@ public final class OffsetManager implements RecordDispatcherListener {
    * {@inheritDoc}
    */
   @Override
-  public void successfullySentToSubscriber(final KafkaConsumerRecord<?, ?> record) {
+  public void successfullySentToSubscriber(final ConsumerRecord<?, ?> record) {
     commit(record);
   }
 
@@ -121,7 +120,7 @@ public final class OffsetManager implements RecordDispatcherListener {
    * {@inheritDoc}
    */
   @Override
-  public void successfullySentToDeadLetterSink(final KafkaConsumerRecord<?, ?> record) {
+  public void successfullySentToDeadLetterSink(final ConsumerRecord<?, ?> record) {
     commit(record);
   }
 
@@ -129,7 +128,7 @@ public final class OffsetManager implements RecordDispatcherListener {
    * {@inheritDoc}
    */
   @Override
-  public void failedToSendToDeadLetterSink(final KafkaConsumerRecord<?, ?> record, final Throwable ex) {
+  public void failedToSendToDeadLetterSink(final ConsumerRecord<?, ?> record, final Throwable ex) {
     commit(record);
   }
 
@@ -137,11 +136,11 @@ public final class OffsetManager implements RecordDispatcherListener {
    * {@inheritDoc}
    */
   @Override
-  public void recordDiscarded(final KafkaConsumerRecord<?, ?> record) {
+  public void recordDiscarded(final ConsumerRecord<?, ?> record) {
     commit(record);
   }
 
-  private void commit(final KafkaConsumerRecord<?, ?> record) {
+  private void commit(final ConsumerRecord<?, ?> record) {
     // We need to handle the case when the offset tracker was removed from our Map since when partitions are revoked we
     // remove the associated offset tracker, however, we may get a response from the sink for a previously owned
     // partition after a partition has been revoked.
@@ -199,7 +198,7 @@ public final class OffsetManager implements RecordDispatcherListener {
    *
    * @return succeeded or failed future.
    */
-  private Future<Void> commit(final Set<TopicPartition> partitions) {
+  private Future<Void> commit(final Collection<TopicPartition> partitions) {
     return CompositeFuture.all(
       this.offsetTrackers.entrySet()
         .stream()
@@ -312,7 +311,7 @@ public final class OffsetManager implements RecordDispatcherListener {
   }
 
   private static void logPartitions(final String context,
-                                    final Set<TopicPartition> tps) {
+                                    final Collection<TopicPartition> tps) {
     logger.info("Partitions " + context + " {}", keyValue("partitions", tps));
   }
 
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
index 619d618cf..ba426a9e0 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
@@ -25,15 +25,17 @@ import io.github.bucket4j.local.LocalBucketBuilder;
 import io.github.bucket4j.local.SynchronizationStrategy;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
-import io.vertx.kafka.client.common.TopicPartition;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecords;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.TopicPartition;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -95,8 +97,9 @@ public class OrderedConsumerVerticle extends ConsumerVerticle {
 
   @Override
   void startConsumer(Promise<Void> startPromise) {
+    Objects.requireNonNull(getConsumerVerticleContext().getConsumerRebalanceListener());
     // We need to sub first, then we can start the polling loop
-    this.consumer.subscribe(Set.copyOf(getConsumerVerticleContext().getResource().getTopicsList()))
+    this.consumer.subscribe(Set.copyOf(getConsumerVerticleContext().getResource().getTopicsList()),getConsumerVerticleContext().getConsumerRebalanceListener())
       .onFailure(startPromise::fail)
       .onSuccess(v -> {
         if (this.pollTimer.compareAndSet(-1, 0)) {
@@ -157,34 +160,33 @@ public class OrderedConsumerVerticle extends ConsumerVerticle {
   }
 
   @Override
-  public PartitionRevokedHandler getPartitionsRevokedHandler() {
+  public PartitionRevokedHandler getPartitionRevokedHandler() {
     return partitionRevokedHandler;
   }
 
-  private void recordsHandler(KafkaConsumerRecords<Object, CloudEvent> records) {
+  private void recordsHandler(ConsumerRecords<Object, CloudEvent> records) {
     if (this.closed.get()) {
       return;
     }
     isPollInFlight.set(false);
 
-    if (records == null || records.size() == 0) {
+    if (records == null || records.count() == 0) {
       return;
     }
 
     if (bucket != null) {
       // Once we have new records, we force add them to internal per-partition queues.
-      bucket.forceAddTokens(records.size());
+      bucket.forceAddTokens(records.count());
     }
 
     // Put records in internal per-partition queues.
-    for (int i = 0; i < records.size(); i++) {
-      final var record = records.recordAt(i);
+    for(var record : records) {
       final var executor = executorFor(new TopicPartition(record.topic(), record.partition()));
       executor.offer(() -> dispatch(record));
     }
   }
 
-  private Future<Void> dispatch(final KafkaConsumerRecord<Object, CloudEvent> record) {
+  private Future<Void> dispatch(final ConsumerRecord<Object, CloudEvent> record) {
     if (this.closed.get()) {
       return Future.failedFuture("Consumer verticle closed " + getConsumerVerticleContext());
     }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/PartitionRevokedHandler.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/PartitionRevokedHandler.java
index bc628e1a9..b6426db15 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/PartitionRevokedHandler.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/PartitionRevokedHandler.java
@@ -16,9 +16,11 @@
 package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
 
 import io.vertx.core.Future;
-import io.vertx.kafka.client.common.TopicPartition;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.Collection;
+
 
-import java.util.Set;
 
 /**
  * {@link PartitionRevokedHandler} is the handler called when some partitions are revoked to a
@@ -30,5 +32,5 @@ public interface PartitionRevokedHandler {
    * @param partitions revoked partitions
    * @return a successful or a failed future
    */
-  Future<Void> partitionRevoked(final Set<TopicPartition> partitions);
+  Future<Void> partitionRevoked(final Collection<TopicPartition> partitions);
 }
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 1e7a0cd6f..e92667c9b 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
@@ -20,12 +20,13 @@ import dev.knative.eventing.kafka.broker.dispatcher.main.ConsumerVerticleContext
 import io.cloudevents.CloudEvent;
 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.apache.kafka.clients.consumer.ConsumerRecords;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -58,9 +59,12 @@ public final class UnorderedConsumerVerticle extends ConsumerVerticle {
 
   @Override
   void startConsumer(final Promise<Void> startPromise) {
-    this.consumer.subscribe(Set.copyOf(getConsumerVerticleContext().getResource().getTopicsList()), startPromise);
-    startPromise.future()
-      .onSuccess(v -> poll());
+    Objects.requireNonNull(getConsumerVerticleContext().getConsumerRebalanceListener());
+
+    this.consumer.subscribe(Set.copyOf(getConsumerVerticleContext().getResource().getTopicsList()), getConsumerVerticleContext().getConsumerRebalanceListener())
+    .onFailure(startPromise::tryFail)
+    .onSuccess(startPromise::tryComplete)
+    .onSuccess(v -> poll());
   }
 
   /**
@@ -103,7 +107,7 @@ public final class UnorderedConsumerVerticle extends ConsumerVerticle {
     }
   }
 
-  private void handleRecords(final KafkaConsumerRecords<Object, CloudEvent> records) {
+  private void handleRecords(final ConsumerRecords<Object, CloudEvent> records) {
     if (closed.get()) {
       isPollInFlight.compareAndSet(true, false);
       return;
@@ -112,16 +116,17 @@ public final class UnorderedConsumerVerticle extends ConsumerVerticle {
     // 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.
-    this.inFlightRecords.addAndGet(records.size());
+    this.inFlightRecords.addAndGet(records.count());
     isPollInFlight.compareAndSet(true, false);
 
-    for (int i = 0; i < records.size(); i++) {
-      this.recordDispatcher.dispatch(records.recordAt(i))
+    for(var record: records){
+      this.recordDispatcher.dispatch(record)
         .onComplete(v -> {
           this.inFlightRecords.decrementAndGet();
           poll();
         });
     }
+    
     poll();
   }
 
@@ -133,7 +138,7 @@ public final class UnorderedConsumerVerticle extends ConsumerVerticle {
   }
 
   @Override
-  public PartitionRevokedHandler getPartitionsRevokedHandler() {
+  public PartitionRevokedHandler getPartitionRevokedHandler() {
     return partitions -> Future.succeededFuture();
   }
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerDeployerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerDeployerVerticle.java
index c79ef2257..ee4e1a9a8 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerDeployerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerDeployerVerticle.java
@@ -24,7 +24,6 @@ import io.vertx.core.DeploymentOptions;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.eventbus.MessageConsumer;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerFactory.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerFactory.java
deleted file mode 100644
index dada0bb0a..000000000
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerFactory.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
- *
- * 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 dev.knative.eventing.kafka.broker.dispatcher.main;
-
-import io.vertx.core.Vertx;
-import io.vertx.core.tracing.TracingPolicy;
-import io.vertx.kafka.client.common.KafkaClientOptions;
-import io.vertx.kafka.client.consumer.KafkaConsumer;
-
-import java.util.Map;
-
-@FunctionalInterface
-public interface ConsumerFactory<K, V> {
-
-  KafkaConsumer<K, V> create(final Vertx vertx, final Map<String, Object> consumerConfigs);
-
-  static <K, V> ConsumerFactory<K, V> defaultFactory() {
-    return new ConsumerFactory<>() {
-
-      private KafkaConsumer<K, V> consumer;
-
-      @Override
-      public KafkaConsumer<K, V> create(Vertx vertx, Map<String, Object> consumerConfigs) {
-        if (consumer == null) {
-          consumer = KafkaConsumer.create(vertx, new KafkaClientOptions().setConfig(consumerConfigs)
-            // Disable tracing provided by vertx-kafka-client, because it doesn't work well with our dispatch logic.
-            // RecordDispatcher, when receiving a new record, takes care of adding the proper receive record span.
-            .setTracingPolicy(TracingPolicy.IGNORE));
-        }
-        return consumer;
-      }
-    };
-  }
-}
-
-
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleBuilder.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleBuilder.java
index 5de9f7dc9..92a754be3 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleBuilder.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleBuilder.java
@@ -22,6 +22,7 @@ import dev.knative.eventing.kafka.broker.core.security.KafkaClientsAuth;
 import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSender;
 import dev.knative.eventing.kafka.broker.dispatcher.DeliveryOrder;
 import dev.knative.eventing.kafka.broker.dispatcher.Filter;
+import dev.knative.eventing.kafka.broker.dispatcher.ReactiveKafkaConsumer;
 import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.NoopResponseHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.RecordDispatcherImpl;
@@ -44,21 +45,20 @@ import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi
 import dev.knative.eventing.kafka.broker.dispatcher.impl.http.WebClientCloudEventSender;
 import io.cloudevents.CloudEvent;
 import io.vertx.core.Future;
-import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.tracing.TracingPolicy;
 import io.vertx.ext.web.client.WebClient;
 import io.vertx.kafka.client.common.KafkaClientOptions;
-import io.vertx.kafka.client.common.TopicPartition;
 import io.vertx.kafka.client.common.tracing.ConsumerTracer;
-import io.vertx.kafka.client.consumer.KafkaConsumer;
 import io.vertx.kafka.client.producer.KafkaProducer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.common.TopicPartition;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -89,7 +89,7 @@ public class ConsumerVerticleBuilder {
     KafkaClientsAuth.attachCredentials(consumerVerticleContext.getConsumerConfigs(), credentials);
     KafkaClientsAuth.attachCredentials(consumerVerticleContext.getProducerConfigs(), credentials);
 
-    final KafkaConsumer<Object, CloudEvent> consumer = this.consumerVerticleContext
+    final ReactiveKafkaConsumer<Object, CloudEvent> consumer = this.consumerVerticleContext
       .getConsumerFactory()
       .create(vertx, consumerVerticleContext.getConsumerConfigs());
     consumerVerticle.setConsumer(consumer);
@@ -127,10 +127,10 @@ public class ConsumerVerticleBuilder {
     consumerVerticle.setRecordDispatcher(recordDispatcher);
 
     final var partitionRevokedHandlers = List.of(
-      consumerVerticle.getPartitionsRevokedHandler(),
+      consumerVerticle.getPartitionRevokedHandler(),
       offsetManager.getPartitionRevokedHandler()
     );
-    consumer.partitionsRevokedHandler(handlePartitionsRevoked(partitionRevokedHandlers));
+    this.consumerVerticleContext.withConsumerRebalanceListener(createRebalanceListener(partitionRevokedHandlers));
   }
 
   private ConsumerVerticle createConsumerVerticle(final ConsumerVerticle.Initializer initializer) {
@@ -140,35 +140,45 @@ public class ConsumerVerticleBuilder {
     };
   }
 
-  /**
+    /**
    * For each handler call partitionRevoked and wait for the future to complete.
    *
    * @param partitionRevokedHandlers partition revoked handlers
-   * @return a single handler that dispatches the partition revoked event to the given handlers.
+   * @return ConsumerRebalanceListener object with the partition revoked handler running on onPartitionsRevoked
    */
-  private Handler<Set<TopicPartition>> handlePartitionsRevoked(final List<PartitionRevokedHandler> partitionRevokedHandlers) {
-    return partitions -> {
-
-      ConsumerVerticleContext.logger.info("Received revoke partitions for consumer {} {}",
-        consumerVerticleContext.getLoggingKeyValue(),
-        keyValue("partitions", partitions)
-      );
+  private ConsumerRebalanceListener createRebalanceListener(final List<PartitionRevokedHandler> partitionRevokedHandlers) {
+    return new ConsumerRebalanceListener() {
+        @Override
+        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+            ConsumerVerticleContext.logger.info("Received revoke partitions for consumer {} {}",
+                    consumerVerticleContext.getLoggingKeyValue(),
+                    keyValue("partitions", partitions)
+            );
+
+            final var futures = new ArrayList<Future<Void>>(partitionRevokedHandlers.size());
+            for (PartitionRevokedHandler partitionRevokedHandler : partitionRevokedHandlers) {
+                futures.add(partitionRevokedHandler.partitionRevoked(partitions));
+            }
+
+            for (final var future : futures) {
+                try {
+                    future.toCompletionStage().toCompletableFuture().get(1, TimeUnit.SECONDS);
+                } catch (final Exception ignored) {
+                    ConsumerVerticleContext.logger.warn("Partition revoked handler failed {} {}",
+                            consumerVerticleContext.getLoggingKeyValue(),
+                            keyValue("partitions", partitions)
+                    );
+                }
+            }
+        }
 
-      final var futures = new ArrayList<Future<Void>>(partitionRevokedHandlers.size());
-      for (PartitionRevokedHandler partitionRevokedHandler : partitionRevokedHandlers) {
-        futures.add(partitionRevokedHandler.partitionRevoked(partitions));
-      }
-
-      for (final var future : futures) {
-        try {
-          future.toCompletionStage().toCompletableFuture().get(1, TimeUnit.SECONDS);
-        } catch (final Exception ignored) {
-          ConsumerVerticleContext.logger.warn("Partition revoked handler failed {} {}",
-            consumerVerticleContext.getLoggingKeyValue(),
-            keyValue("partitions", partitions)
-          );
+        @Override
+        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+            ConsumerVerticleContext.logger.info("Received assign partitions for consumer {} {}",
+                    consumerVerticleContext.getLoggingKeyValue(),
+                    keyValue("partitions", partitions)
+            );
         }
-      }
     };
   }
 
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleContext.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleContext.java
index 4b5521226..865b510ce 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleContext.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleContext.java
@@ -18,6 +18,7 @@ package dev.knative.eventing.kafka.broker.dispatcher.main;
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
+import dev.knative.eventing.kafka.broker.dispatcher.ReactiveConsumerFactory;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.InvalidCloudEventInterceptor;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.KeyDeserializer;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.http.WebClientCloudEventSender;
@@ -27,6 +28,7 @@ import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.Tags;
 import io.vertx.ext.web.client.WebClientOptions;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -53,7 +55,7 @@ public class ConsumerVerticleContext {
   private Map<String, Object> producerConfigs;
   private WebClientOptions webClientOptions;
 
-  private ConsumerFactory<Object, CloudEvent> consumerFactory;
+  private ReactiveConsumerFactory<Object, CloudEvent> consumerFactory;
   private ProducerFactory<String, CloudEvent> producerFactory;
 
   private Integer maxPollRecords;
@@ -63,6 +65,8 @@ public class ConsumerVerticleContext {
 
   private Tags tags;
 
+  private ConsumerRebalanceListener consumerRebalanceListener;
+
   public ConsumerVerticleContext withConsumerConfigs(final Map<String, Object> consumerConfigs) {
     this.consumerConfigs = new HashMap<>(consumerConfigs);
     return this;
@@ -137,7 +141,7 @@ public class ConsumerVerticleContext {
     return this;
   }
 
-  public ConsumerVerticleContext withConsumerFactory(final ConsumerFactory<Object, CloudEvent> consumerFactory) {
+  public ConsumerVerticleContext withConsumerFactory(final ReactiveConsumerFactory<Object, CloudEvent> consumerFactory) {
     this.consumerFactory = consumerFactory;
     return this;
   }
@@ -147,6 +151,15 @@ public class ConsumerVerticleContext {
     return this;
   }
 
+  public ConsumerVerticleContext withConsumerRebalanceListener(final ConsumerRebalanceListener consumerRebalanceListener) {
+    this.consumerRebalanceListener = consumerRebalanceListener;
+    return this;
+  }
+
+  public ConsumerRebalanceListener getConsumerRebalanceListener() {
+    return consumerRebalanceListener;
+  }
+   
   public DataPlaneContract.Resource getResource() {
     return resource;
   }
@@ -202,7 +215,7 @@ public class ConsumerVerticleContext {
     return keyValue("context", getLoggingContext());
   }
 
-  public ConsumerFactory<Object, CloudEvent> getConsumerFactory() {
+  public ReactiveConsumerFactory<Object, CloudEvent> getConsumerFactory() {
     return this.consumerFactory;
   }
 
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 ac0c22e88..6865a5489 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
@@ -18,6 +18,7 @@ package dev.knative.eventing.kafka.broker.dispatcher.main;
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
 import dev.knative.eventing.kafka.broker.dispatcher.ConsumerVerticleFactory;
+import dev.knative.eventing.kafka.broker.dispatcher.ReactiveConsumerFactory;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.ConsumerVerticle;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.vertx.ext.web.client.WebClientOptions;
@@ -36,6 +37,7 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
   private final Map<String, Object> producerConfigs;
   private final AuthProvider authProvider;
   private final MeterRegistry metricsRegistry;
+  private final ReactiveConsumerFactory reactiveConsumerFactory;
 
   /**
    * All args constructor.
@@ -51,7 +53,8 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
     final WebClientOptions webClientOptions,
     final Properties producerConfigs,
     final AuthProvider authProvider,
-    final MeterRegistry metricsRegistry) {
+    final MeterRegistry metricsRegistry,
+    final ReactiveConsumerFactory reactiveConsumerFactory) {
 
     Objects.requireNonNull(consumerConfigs, "provide consumerConfigs");
     Objects.requireNonNull(webClientOptions, "provide webClientOptions");
@@ -69,6 +72,7 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
     this.webClientOptions = webClientOptions;
     this.authProvider = authProvider;
     this.metricsRegistry = metricsRegistry;
+    this.reactiveConsumerFactory = reactiveConsumerFactory;
   }
 
   /**
@@ -87,7 +91,7 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
         .withAuthProvider(authProvider)
         .withMeterRegistry(metricsRegistry)
         .withResource(resource, egress)
-        .withConsumerFactory(ConsumerFactory.defaultFactory())
+        .withConsumerFactory(reactiveConsumerFactory)
         .withProducerFactory(ProducerFactory.defaultFactory())
     )
       .build();
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/Main.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/Main.java
index 2ea948cb9..5815aabb4 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/Main.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/Main.java
@@ -24,6 +24,7 @@ import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
 import dev.knative.eventing.kafka.broker.core.tracing.TracingConfig;
 import dev.knative.eventing.kafka.broker.core.utils.Configurations;
 import dev.knative.eventing.kafka.broker.core.utils.Shutdown;
+import dev.knative.eventing.kafka.broker.dispatcher.ReactiveConsumerFactory;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.CloudEventDeserializer;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.InvalidCloudEventInterceptor;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.KeyDeserializer;
@@ -63,7 +64,7 @@ public class Main {
    *
    * @param args command line arguments.
    */
-  public static void main(final String[] args) throws IOException {
+  public static void start(final String[] args, final ReactiveConsumerFactory reactiveConsumerFactory) throws IOException {
     DispatcherEnv env = new DispatcherEnv(System::getenv);
 
     OpenTelemetrySdk openTelemetry = TracingConfig.fromDir(env.getConfigTracingPath()).setup();
@@ -108,7 +109,8 @@ public class Main {
           webClientOptions,
           producerConfig,
           AuthProvider.kubernetes(),
-          Metrics.getRegistry()
+          Metrics.getRegistry(),
+          reactiveConsumerFactory
         ),
         env.getEgressesInitialCapacity()
       );
diff --git a/data-plane/profiler/run.sh b/data-plane/profiler/run.sh
index 233895c1c..46a61b360 100755
--- a/data-plane/profiler/run.sh
+++ b/data-plane/profiler/run.sh
@@ -126,7 +126,7 @@ java \
   -XX:+UnlockDiagnosticVMOptions \
   -XX:+DebugNonSafepoints \
   -Dlogback.configurationFile="${RESOURCES_DIR}"/config-logging.xml \
-  -jar "${PROJECT_ROOT_DIR}"/dispatcher/target/dispatcher-1.0-SNAPSHOT.jar >"${LOG_DIR}/dispatcher.log" &
+  -jar "${PROJECT_ROOT_DIR}"/dispatcher-vertx/target/dispatcher-vertx-1.0-SNAPSHOT.jar >"${LOG_DIR}/dispatcher.log" &
 dispatcher_pid=$!
 
 # Download Sacura
diff --git a/hack/data-plane.sh b/hack/data-plane.sh
index 2290504fe..73463521f 100755
--- a/hack/data-plane.sh
+++ b/hack/data-plane.sh
@@ -32,7 +32,7 @@ readonly CHANNEL_DATA_PLANE_CONFIG_DIR=${DATA_PLANE_CONFIG_DIR}/channel
 readonly CHANNEL_TLS_CONFIG_DIR=${DATA_PLANE_CONFIG_DIR}/channel-tls
 
 readonly RECEIVER_DIRECTORY=receiver-vertx
-readonly DISPATCHER_DIRECTORY=dispatcher
+readonly DISPATCHER_DIRECTORY=dispatcher-vertx
 
 # Checks whether the given function exists.
 function function_exists() {


Test patch:
diff --git a/control-plane/pkg/prober/composite_prober_test.go b/control-plane/pkg/prober/composite_prober_test.go
new file mode 100644
index 000000000..2f8cb3cce
--- /dev/null
+++ b/control-plane/pkg/prober/composite_prober_test.go
@@ -0,0 +1,231 @@
+/*
+ * Copyright 2023 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 prober
+
+import (
+	"context"
+	"crypto/tls"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/pointer"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+	"knative.dev/pkg/network"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+)
+
+func TestCompositeProber(t *testing.T) {
+	t.Parallel()
+
+	tt := []struct {
+		name                     string
+		pods                     []*corev1.Pod
+		podsLabelsSelector       labels.Selector
+		addressable              NewAddressable
+		responseStatusCode       int
+		wantStatus               Status
+		wantRequeueCountMin      int
+		wantHttpRequestCountMin  int
+		wantHttpsRequestCountMin int
+	}{
+		{
+			name: "one pod - http only",
+			pods: []*corev1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns",
+						Name:      "p1",
+						Labels:    map[string]string{"app": "p"},
+					},
+					Status: corev1.PodStatus{PodIP: "127.0.0.1"},
+				},
+			},
+			podsLabelsSelector: labels.SelectorFromSet(map[string]string{"app": "p"}),
+			addressable: NewAddressable{
+				AddressStatus: &duckv1.AddressStatus{
+					Address: &duckv1.Addressable{
+						URL: &apis.URL{Scheme: "http", Path: "/b1/b1"},
+					},
+					Addresses: []duckv1.Addressable{
+						{
+							URL: &apis.URL{Scheme: "http", Path: "/b1/b1"},
+						},
+					},
+				},
+			},
+			responseStatusCode:       http.StatusOK,
+			wantStatus:               StatusReady,
+			wantRequeueCountMin:      1,
+			wantHttpRequestCountMin:  1,
+			wantHttpsRequestCountMin: 0,
+		},
+		{
+			name: "one pod - https only",
+			pods: []*corev1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns",
+						Name:      "p1",
+						Labels:    map[string]string{"app": "p"},
+					},
+					Status: corev1.PodStatus{PodIP: "127.0.0.1"},
+				},
+			},
+			podsLabelsSelector: labels.SelectorFromSet(map[string]string{"app": "p"}),
+			addressable: NewAddressable{
+				AddressStatus: &duckv1.AddressStatus{
+					Address: &duckv1.Addressable{
+						URL: &apis.URL{Scheme: "https", Path: "/b1/b1"},
+					},
+					Addresses: []duckv1.Addressable{
+						{
+							URL: &apis.URL{Scheme: "https", Path: "/b1/b1"},
+						},
+					},
+				},
+			},
+			responseStatusCode:       http.StatusOK,
+			wantStatus:               StatusReady,
+			wantRequeueCountMin:      1,
+			wantHttpRequestCountMin:  0,
+			wantHttpsRequestCountMin: 1,
+		},
+		{
+			name: "one pod - http and https, http primary address",
+			pods: []*corev1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns",
+						Name:      "p1",
+						Labels:    map[string]string{"app": "p"},
+					},
+					Status: corev1.PodStatus{PodIP: "127.0.0.1"},
+				},
+			},
+			podsLabelsSelector: labels.SelectorFromSet(map[string]string{"app": "p"}),
+			addressable: NewAddressable{
+				AddressStatus: &duckv1.AddressStatus{
+					Address: &duckv1.Addressable{
+						URL: &apis.URL{Scheme: "http", Path: "/b1/b1"},
+					},
+					Addresses: []duckv1.Addressable{
+						{
+							URL: &apis.URL{Scheme: "http", Path: "/b1/b1"},
+						},
+						{
+							URL: &apis.URL{Scheme: "https", Path: "/b1/b1"},
+						},
+					},
+				},
+			},
+			responseStatusCode:       http.StatusOK,
+			wantStatus:               StatusReady,
+			wantRequeueCountMin:      2,
+			wantHttpRequestCountMin:  1,
+			wantHttpsRequestCountMin: 1,
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx, _ := reconcilertesting.SetupFakeContext(t)
+			ctx, cancel := context.WithCancel(ctx)
+			defer func() {
+				time.Sleep(time.Second)
+				cancel()
+			}()
+			wantHttpRequestCountMin := atomic.NewInt64(int64(tc.wantHttpRequestCountMin))
+			httpHandler := http.HandlerFunc(func(writer http.ResponseWriter, r *http.Request) {
+				wantHttpRequestCountMin.Dec()
+				require.Equal(t, network.ProbeHeaderValue, r.Header.Get(network.ProbeHeaderName))
+				writer.WriteHeader(tc.responseStatusCode)
+			})
+			httpServer := httptest.NewUnstartedServer(httpHandler)
+			httpServer.Start()
+			defer httpServer.Close()
+
+			wantHttpsRequestCountMin := atomic.NewInt64(int64(tc.wantHttpsRequestCountMin))
+			httpsHandler := http.HandlerFunc(func(writer http.ResponseWriter, r *http.Request) {
+				wantHttpsRequestCountMin.Dec()
+				require.Equal(t, network.ProbeHeaderValue, r.Header.Get(network.ProbeHeaderName))
+				writer.WriteHeader(tc.responseStatusCode)
+			})
+			httpsServer := httptest.NewUnstartedServer(httpsHandler)
+			cert, err := tls.X509KeyPair(Crt1, Key1)
+			require.NoError(t, err)
+			httpsServer.TLS = &tls.Config{
+				Certificates: []tls.Certificate{cert},
+			}
+			httpsServer.StartTLS()
+			defer httpsServer.Close()
+
+			for _, p := range tc.pods {
+				_ = podinformer.Get(ctx).Informer().GetStore().Add(p)
+			}
+
+			for _, addr := range tc.addressable.AddressStatus.Addresses {
+				if addr.URL.Scheme == "http" {
+					addr.URL.Host = httpServer.URL
+				} else {
+					addr.URL.Host = httpsServer.URL
+				}
+			}
+
+			var IPsLister IPsLister = func(addressable Addressable) ([]string, error) {
+				pods, err := podinformer.Get(ctx).Lister().List(tc.podsLabelsSelector)
+				if err != nil {
+					return nil, err
+				}
+				ips := make([]string, 0, len(pods))
+				for _, p := range pods {
+					ips = append(ips, p.Status.PodIP)
+				}
+				return ips, nil
+			}
+
+			httpUrl, _ := url.Parse(httpServer.URL)
+			httpsUrl, _ := url.Parse(httpsServer.URL)
+			wantRequeueCountMin := atomic.NewInt64(int64(tc.wantRequeueCountMin))
+			prober, err := NewComposite(ctx, httpUrl.Port(), httpsUrl.Port(), IPsLister, func(key types.NamespacedName) {
+				wantRequeueCountMin.Dec()
+			}, pointer.String(string(CA1)))
+			require.NoError(t, err)
+
+			probeFunc := func() bool {
+				status := prober.Probe(ctx, tc.addressable, tc.wantStatus)
+				return status == tc.wantStatus
+			}
+
+			require.Eventuallyf(t, probeFunc, 5*time.Second, 250*time.Millisecond, "")
+			require.Eventuallyf(t, func() bool { return wantHttpRequestCountMin.Load() == 0 }, 5*time.Second, 250*time.Millisecond, "got %d, want 0", wantHttpRequestCountMin.Load())
+			require.Eventuallyf(t, func() bool { return wantHttpsRequestCountMin.Load() == 0 }, 5*time.Second, 250*time.Millisecond, "got %d, want 0", wantHttpsRequestCountMin.Load())
+			require.Eventuallyf(t, func() bool { return wantRequeueCountMin.Load() == 0 }, 5*time.Second, 250*time.Millisecond, "got %d, want 0", wantRequeueCountMin.Load())
+
+		})
+	}
+}
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutorTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutorTest.java
index d3829a8d1..d49a731da 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutorTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutorTest.java
@@ -30,7 +30,7 @@ import java.util.function.Supplier;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
-import io.vertx.kafka.client.common.TopicPartition;
+import org.apache.kafka.common.TopicPartition;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/MockReactiveConsumerFactory.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/MockReactiveConsumerFactory.java
new file mode 100644
index 000000000..56ad1bac5
--- /dev/null
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/MockReactiveConsumerFactory.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * 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 dev.knative.eventing.kafka.broker.dispatcher;
+
+import java.util.Map;
+
+import io.vertx.core.Vertx;
+
+public class MockReactiveConsumerFactory<K, V> implements ReactiveConsumerFactory<K, V> {
+
+    @Override
+    public ReactiveKafkaConsumer<K, V> create(Vertx vertx, Map<String, Object> configs) {
+        return new MockReactiveKafkaConsumer<>(configs);
+    }
+    
+}
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/MockReactiveKafkaConsumer.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/MockReactiveKafkaConsumer.java
new file mode 100644
index 000000000..a0f9b8e6c
--- /dev/null
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/MockReactiveKafkaConsumer.java
@@ -0,0 +1,102 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * 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 dev.knative.eventing.kafka.broker.dispatcher;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
+import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
+import io.vertx.core.impl.ContextInternal;
+
+public class MockReactiveKafkaConsumer<K, V> implements ReactiveKafkaConsumer<K, V> {
+
+    private Consumer<K, V> consumer;
+
+    public MockReactiveKafkaConsumer(Map<String, Object> configs) {
+        consumer = new KafkaConsumer<K, V>(configs);
+    }
+
+    public MockReactiveKafkaConsumer(Consumer<K, V> consumer) {
+        this.consumer = consumer;
+    }
+
+    @Override
+    public Future<Map<TopicPartition, OffsetAndMetadata>> commit(Map<TopicPartition, OffsetAndMetadata> offset) {
+        consumer.commitSync(offset);
+        return Future.succeededFuture(offset);
+    }
+
+    @Override
+    public Future<Void> close() {
+        consumer.close();
+        return Future.succeededFuture();
+    }
+
+    @Override
+    public Future<Void> pause(Collection<TopicPartition> partitions) {
+        consumer.pause(partitions);
+        return Future.succeededFuture();
+    }
+
+    @Override
+    public Future<ConsumerRecords<K, V>> poll(Duration timeout) {
+        final Promise<ConsumerRecords<K, V>> p = Promise.promise();
+        final var records = consumer.poll(timeout);
+        
+        // waits timeout time for new records to be available.
+        ((ContextInternal) Vertx.currentContext()).setTimer(timeout.toMillis(), v -> p.complete(records));
+        return p.future();
+    }
+
+    @Override
+    public Future<Void> resume(Collection<TopicPartition> partitions) {
+        consumer.resume(partitions);
+        return Future.succeededFuture();
+    }
+
+    @Override
+    public Future<Void> subscribe(Collection<String> topics) {
+        consumer.subscribe(topics);
+        return Future.succeededFuture();
+    }
+
+    @Override
+    public Future<Void> subscribe(Collection<String> topics, ConsumerRebalanceListener listener) {
+        consumer.subscribe(topics, listener);
+        return Future.succeededFuture();
+    }
+
+    @Override
+    public Consumer<K, V> unwrap() {
+        return this.consumer;
+    }
+
+    @Override
+    public ReactiveKafkaConsumer<K, V> exceptionHandler(Handler<Throwable> handler) {
+        return this;
+    }
+}
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChainTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChainTest.java
index 5f7a31614..93cbdca6b 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChainTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherMutatorChainTest.java
@@ -19,7 +19,6 @@ import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.builder.CloudEventBuilder;
 import io.vertx.core.Future;
-import io.vertx.kafka.client.consumer.impl.KafkaConsumerRecordImpl;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.junit.jupiter.api.Test;
 
@@ -53,16 +52,15 @@ public class RecordDispatcherMutatorChainTest {
     );
     when(next.dispatch(any())).thenReturn(Future.succeededFuture());
 
-    final var givenRecord = new KafkaConsumerRecordImpl<>(
-      new ConsumerRecord<>(
+    final var givenRecord = new ConsumerRecord<>(
         "t1",
         0,
         0,
         (Object) "abc",
         CloudEventBuilder.from(expected)
           .withId(UUID.randomUUID().toString())
-          .build())
-    );
+          .build()
+      );
 
     final var succeeded = chain.dispatch(givenRecord);
 
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 71743bc33..145b0b98c 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
@@ -39,8 +39,6 @@ import io.vertx.core.http.HttpVersion;
 import io.vertx.ext.web.client.impl.HttpResponseImpl;
 import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
-import io.vertx.kafka.client.consumer.impl.KafkaConsumerRecordImpl;
 import io.vertx.micrometer.MicrometerMetricsOptions;
 import io.vertx.micrometer.backends.BackendRegistries;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -280,7 +278,7 @@ public class RecordDispatcherTest {
     final var record = record();
     dispatcherHandler.dispatch(record);
 
-    ArgumentCaptor<KafkaConsumerRecord<Object, CloudEvent>> captor = ArgumentCaptor.forClass(KafkaConsumerRecord.class);
+    ArgumentCaptor<ConsumerRecord<Object, CloudEvent>> captor = ArgumentCaptor.forClass(ConsumerRecord.class);
 
     assertTrue(subscriberSenderSendCalled.get());
     assertTrue(dlsSenderSendCalled.get());
@@ -290,7 +288,7 @@ public class RecordDispatcherTest {
     verify(receiver, never()).failedToSendToDeadLetterSink(any(), any());
     verify(receiver, never()).recordDiscarded(any());
 
-    KafkaConsumerRecord<Object, CloudEvent> failedRecord = captor.getValue();
+    ConsumerRecord<Object, CloudEvent> failedRecord = captor.getValue();
     assertEquals(record.topic(), failedRecord.topic());
     assertEquals(record.partition(), failedRecord.partition());
     assertEquals(record.offset(), failedRecord.offset());
@@ -343,7 +341,7 @@ public class RecordDispatcherTest {
     final var record = record();
     dispatcherHandler.dispatch(record);
 
-    ArgumentCaptor<KafkaConsumerRecord<Object, CloudEvent>> captor = ArgumentCaptor.forClass(KafkaConsumerRecord.class);
+    ArgumentCaptor<ConsumerRecord<Object, CloudEvent>> captor = ArgumentCaptor.forClass(ConsumerRecord.class);
 
     assertTrue(subscriberSenderSendCalled.get());
     assertTrue(dlsSenderSendCalled.get());
@@ -353,7 +351,7 @@ public class RecordDispatcherTest {
     verify(receiver, never()).failedToSendToDeadLetterSink(any(), any());
     verify(receiver, never()).recordDiscarded(any());
 
-    KafkaConsumerRecord<Object, CloudEvent> failedRecord = captor.getValue();
+    ConsumerRecord<Object, CloudEvent> failedRecord = captor.getValue();
     assertEquals(record.topic(), failedRecord.topic());
     assertEquals(record.partition(), failedRecord.partition());
     assertEquals(record.offset(), failedRecord.offset());
@@ -406,7 +404,7 @@ public class RecordDispatcherTest {
     final var record = record();
     dispatcherHandler.dispatch(record);
 
-    ArgumentCaptor<KafkaConsumerRecord<Object, CloudEvent>> captor = ArgumentCaptor.forClass(KafkaConsumerRecord.class);
+    ArgumentCaptor<ConsumerRecord<Object, CloudEvent>> captor = ArgumentCaptor.forClass(ConsumerRecord.class);
 
     assertTrue(subscriberSenderSendCalled.get());
     assertTrue(dlsSenderSendCalled.get());
@@ -416,7 +414,7 @@ public class RecordDispatcherTest {
     verify(receiver, never()).failedToSendToDeadLetterSink(any(), any());
     verify(receiver, never()).recordDiscarded(any());
 
-    KafkaConsumerRecord<Object, CloudEvent> failedRecord = captor.getValue();
+    ConsumerRecord<Object, CloudEvent> failedRecord = captor.getValue();
     assertEquals(record.topic(), failedRecord.topic());
     assertEquals(record.partition(), failedRecord.partition());
     assertEquals(record.offset(), failedRecord.offset());
@@ -550,13 +548,13 @@ public class RecordDispatcherTest {
     assertNoEventDispatchLatency();
   }
 
-  private static KafkaConsumerRecord<Object, CloudEvent> record() {
-    return new KafkaConsumerRecordImpl<>(new ConsumerRecord<>("", 0, 0L, "", CoreObjects.event()));
+  private static ConsumerRecord<Object, CloudEvent> record() {
+    return new ConsumerRecord<>("", 0, 0L, "", CoreObjects.event());
   }
 
 
-  private static KafkaConsumerRecord<Object, CloudEvent> invalidRecord() {
-    return new KafkaConsumerRecordImpl<>(new ConsumerRecord<>("", 0, 0L, "", new InvalidCloudEvent(new byte[]{1, 4})));
+  private static ConsumerRecord<Object, CloudEvent> invalidRecord() {
+    return new ConsumerRecord<>("", 0, 0L, "", new InvalidCloudEvent(new byte[]{1, 4}));
   }
 
   public static RecordDispatcherListener offsetManagerMock() {
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java
index 345e81e51..df5c2a094 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractConsumerVerticleTest.java
@@ -19,22 +19,22 @@ import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSender;
 import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSenderMock;
+import dev.knative.eventing.kafka.broker.dispatcher.MockReactiveKafkaConsumer;
 import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandlerMock;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.RecordDispatcherImpl;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.RecordDispatcherTest;
 import dev.knative.eventing.kafka.broker.dispatcher.main.ConsumerVerticleContext;
 import dev.knative.eventing.kafka.broker.dispatcher.main.FakeConsumerVerticleContext;
 import io.cloudevents.CloudEvent;
-import io.vertx.core.AsyncResult;
 import io.vertx.core.Future;
-import io.vertx.core.Handler;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
-import io.vertx.kafka.client.consumer.KafkaConsumer;
 import io.vertx.micrometer.MicrometerMetricsOptions;
 import io.vertx.micrometer.backends.BackendRegistries;
+
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.MockConsumer;
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
 import org.apache.kafka.common.TopicPartition;
@@ -49,7 +49,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.*;
-import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.resource1;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.any;
@@ -90,7 +89,7 @@ public abstract class AbstractConsumerVerticleTest {
     final var verticle = createConsumerVerticle(
       consumerVerticleContext,
       (vx, consumerVerticle) -> {
-        consumerVerticle.setConsumer(KafkaConsumer.create(vx, consumer));
+        consumerVerticle.setConsumer(new MockReactiveKafkaConsumer<>(consumer));
         consumerVerticle.setRecordDispatcher(recordDispatcher);
         consumerVerticle.setCloser(Future::succeededFuture);
 
@@ -131,7 +130,7 @@ public abstract class AbstractConsumerVerticleTest {
         egress1()
       ),
       (vx, consumerVerticle) -> {
-        consumerVerticle.setConsumer(KafkaConsumer.create(vx, consumer));
+        consumerVerticle.setConsumer(new MockReactiveKafkaConsumer<>(consumer));
         consumerVerticle.setRecordDispatcher(recordDispatcher);
         consumerVerticle.setCloser(Future::succeededFuture);
 
@@ -161,16 +160,12 @@ public abstract class AbstractConsumerVerticleTest {
   @SuppressWarnings("unchecked")
   public void shouldCloseEverything(final Vertx vertx, final VertxTestContext context) {
     final var topics = new String[]{"a"};
-    final KafkaConsumer<Object, CloudEvent> consumer = mock(KafkaConsumer.class);
+    final MockReactiveKafkaConsumer<Object, CloudEvent> consumer = mock(MockReactiveKafkaConsumer.class);
 
     final var checkpoints = context.checkpoint(2);
 
     when(consumer.close()).thenReturn(Future.succeededFuture());
-    when(consumer.subscribe((Set<String>) any(), any())).then(answer -> {
-      final Handler<AsyncResult<Void>> callback = answer.getArgument(1);
-      callback.handle(Future.succeededFuture());
-      return consumer;
-    });
+    when(consumer.subscribe((Set<String>) any(), any(ConsumerRebalanceListener.class))).thenReturn(Future.succeededFuture());
     when(consumer.subscribe(any(Set.class))).thenReturn(Future.succeededFuture());
     when(consumer.poll(any()))
       .then(answer -> {
@@ -187,7 +182,7 @@ public abstract class AbstractConsumerVerticleTest {
       mockConsumer.unsubscribe();
 
       mockConsumer.assign(Arrays.stream(topics)
-        .map(t -> new TopicPartition(t, 0))
+        .map(topic -> new TopicPartition(topic, 0))
         .collect(Collectors.toSet())
       );
     });
@@ -245,8 +240,9 @@ public abstract class AbstractConsumerVerticleTest {
         }))
       );
 
-
-    await().untilAsserted(() -> verify(consumer, times(1)).close());
+    await().untilAsserted(() -> {
+      verify(consumer, times(1)).close();
+    });
     checkpoints.flag();
   }
 
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractOffsetManagerTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractOffsetManagerTest.java
index e8136d32c..b08504dfd 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractOffsetManagerTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractOffsetManagerTest.java
@@ -15,15 +15,11 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
 
+import dev.knative.eventing.kafka.broker.dispatcher.ReactiveKafkaConsumer;
 import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import io.cloudevents.CloudEvent;
 import io.vertx.core.Future;
-import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
-import io.vertx.kafka.client.consumer.KafkaConsumer;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
-import io.vertx.kafka.client.consumer.OffsetAndMetadata;
-import io.vertx.kafka.client.consumer.impl.KafkaConsumerRecordImpl;
 
 import java.util.Collection;
 import java.util.Map;
@@ -35,6 +31,7 @@ import java.util.stream.Collectors;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
 import org.apache.kafka.common.TopicPartition;
 import org.assertj.core.api.MapAssert;
@@ -48,17 +45,15 @@ import static org.mockito.Mockito.verify;
 
 public abstract class AbstractOffsetManagerTest {
 
-  abstract RecordDispatcherListener createOffsetManager(final Vertx vertx, final KafkaConsumer<?, ?> consumer);
+  abstract RecordDispatcherListener createOffsetManager(final Vertx vertx, final ReactiveKafkaConsumer<?, ?> consumer);
 
-  protected static KafkaConsumerRecord<String, CloudEvent> record(String topic, int partition, long offset) {
-    return new KafkaConsumerRecordImpl<>(
-      new ConsumerRecord<>(
+  protected static ConsumerRecord<String, CloudEvent> record(String topic, int partition, long offset) {
+    return new ConsumerRecord<>(
         topic,
         partition,
         offset,
         null,
         null
-      )
     );
   }
 
@@ -78,23 +73,15 @@ public abstract class AbstractOffsetManagerTest {
     // Funky flag to flip in order to induce a failure
     AtomicBoolean failureFlag = new AtomicBoolean(false);
 
-    final KafkaConsumer<String, CloudEvent> vertxConsumer = mock(KafkaConsumer.class);
+    final ReactiveKafkaConsumer<String, CloudEvent> vertxConsumer = mock(ReactiveKafkaConsumer.class);
     doAnswer(invocation -> {
       if (failureFlag.get()) {
         return Future.failedFuture("some failure");
       }
       // If you don't want to lose hours in debugging, please don't remove this FQCNs :)
-      final Map<io.vertx.kafka.client.common.TopicPartition, OffsetAndMetadata>
+      final Map<TopicPartition, OffsetAndMetadata>
         topicsPartitions = invocation.getArgument(0);
-      mockConsumer.commitSync(
-        topicsPartitions.entrySet()
-          .stream()
-          .collect(Collectors.toMap(
-            e -> new TopicPartition(e.getKey().getTopic(), e.getKey().getPartition()),
-            e -> new org.apache.kafka.clients.consumer.OffsetAndMetadata(e.getValue().getOffset(),
-              e.getValue().getMetadata())
-          ))
-      );
+        mockConsumer.commitSync(topicsPartitions);
       return Future.succeededFuture();
     })
       .when(vertxConsumer)
@@ -120,19 +107,12 @@ public abstract class AbstractOffsetManagerTest {
   }
 
   @SuppressWarnings("unchecked")
-  protected void shouldNeverCommit(final KafkaConsumer<String, CloudEvent> consumer) {
-    verify(consumer, never()).commit();
-    verify(consumer, never()).commit(any(Handler.class));
+  protected void shouldNeverCommit(final ReactiveKafkaConsumer<String, CloudEvent> consumer) {
     verify(consumer, never()).commit(any(Map.class));
-    verify(consumer, never()).commit(any(), any());
   }
 
   @SuppressWarnings("unchecked")
-  protected void shouldNeverPause(final KafkaConsumer<String, CloudEvent> consumer) {
-    verify(consumer, never()).pause();
-    verify(consumer, never()).pause(any(io.vertx.kafka.client.common.TopicPartition.class));
-    verify(consumer, never()).pause(any(Set.class));
-    verify(consumer, never()).pause(any(io.vertx.kafka.client.common.TopicPartition.class), any());
-    verify(consumer, never()).pause(any(Set.class), any());
+  protected void shouldNeverPause(final ReactiveKafkaConsumer<String, CloudEvent> consumer) {
+    verify(consumer, never()).pause(any(Collection.class));
   }
 }
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManagerTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManagerTest.java
index 9660745d9..58a241154 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManagerTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManagerTest.java
@@ -15,14 +15,15 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
 
+import dev.knative.eventing.kafka.broker.dispatcher.ReactiveKafkaConsumer;
 import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import io.cloudevents.CloudEvent;
 import io.micrometer.core.instrument.Counter;
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
 import io.vertx.junit5.VertxExtension;
-import io.vertx.kafka.client.consumer.KafkaConsumer;
-import io.vertx.kafka.client.consumer.OffsetAndMetadata;
+
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
@@ -49,7 +50,7 @@ import static org.mockito.Mockito.when;
 public class OffsetManagerTest extends AbstractOffsetManagerTest {
 
   @Override
-  RecordDispatcherListener createOffsetManager(final Vertx vertx, final KafkaConsumer<?, ?> consumer) {
+  RecordDispatcherListener createOffsetManager(final Vertx vertx, final ReactiveKafkaConsumer<?, ?> consumer) {
     return new OffsetManager(vertx, consumer, null, 100L);
   }
 
@@ -301,7 +302,7 @@ public class OffsetManagerTest extends AbstractOffsetManagerTest {
   @Test
   @SuppressWarnings("unchecked")
   public void recordReceived(final Vertx vertx) {
-    final KafkaConsumer<String, CloudEvent> consumer = mock(KafkaConsumer.class);
+    final ReactiveKafkaConsumer<String, CloudEvent> consumer = mock(ReactiveKafkaConsumer.class);
     final Counter eventsSentCounter = mock(Counter.class);
     new OffsetManager(vertx, consumer, eventsSentCounter::increment, 100L).recordReceived(record("aaa", 0, 0));
 
@@ -313,7 +314,7 @@ public class OffsetManagerTest extends AbstractOffsetManagerTest {
   @Test
   @SuppressWarnings("unchecked")
   public void failedToSendToDeadLetterSink(final Vertx vertx) {
-    final KafkaConsumer<String, CloudEvent> consumer = mock(KafkaConsumer.class);
+    final ReactiveKafkaConsumer<String, CloudEvent> consumer = mock(ReactiveKafkaConsumer.class);
     final Counter eventsSentCounter = mock(Counter.class);
 
     OffsetManager strategy = new OffsetManager(vertx, consumer, eventsSentCounter::increment, 100L);
@@ -331,8 +332,8 @@ public class OffsetManagerTest extends AbstractOffsetManagerTest {
     final var counter = new AtomicInteger(0);
     final Counter eventsSentCounter = mock(Counter.class);
 
-    final KafkaConsumer<String, CloudEvent> consumer = mock(KafkaConsumer.class);
-    when(consumer.commit((Map<io.vertx.kafka.client.common.TopicPartition, OffsetAndMetadata>) any())).then(invocationOnMock -> {
+    final ReactiveKafkaConsumer<String, CloudEvent> consumer = mock(ReactiveKafkaConsumer.class);
+    when(consumer.commit((Map<TopicPartition, OffsetAndMetadata>) any())).then(invocationOnMock -> {
       if (counter.incrementAndGet() == 1) {
         return Future.failedFuture(new RuntimeException());
       }
@@ -347,7 +348,7 @@ public class OffsetManagerTest extends AbstractOffsetManagerTest {
 
     final var offset = strategy.
       getOffsetTrackers().
-      get(new io.vertx.kafka.client.common.TopicPartition(r.topic(), r.partition()));
+      get(new TopicPartition(r.topic(), r.partition()));
 
     await()
       .timeout(Duration.ofSeconds(1))
@@ -360,8 +361,8 @@ public class OffsetManagerTest extends AbstractOffsetManagerTest {
     final var counter = new AtomicInteger(0);
     final Counter eventsSentCounter = mock(Counter.class);
 
-    final KafkaConsumer<String, CloudEvent> consumer = mock(KafkaConsumer.class);
-    when(consumer.commit((Map<io.vertx.kafka.client.common.TopicPartition, OffsetAndMetadata>) any()))
+    final ReactiveKafkaConsumer<String, CloudEvent> consumer = mock(ReactiveKafkaConsumer.class);
+    when(consumer.commit((Map<TopicPartition, OffsetAndMetadata>) any()))
       .then(invocationOnMock -> Future.failedFuture(new RuntimeException()));
 
     final var r = record("aaa", 0, 0);
@@ -372,7 +373,7 @@ public class OffsetManagerTest extends AbstractOffsetManagerTest {
 
     final var offset = strategy.
       getOffsetTrackers().
-      get(new io.vertx.kafka.client.common.TopicPartition(r.topic(), r.partition()));
+      get(new TopicPartition(r.topic(), r.partition()));
 
     await()
       .timeout(Duration.ofSeconds(1))
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticleTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticleTest.java
index 5aca16d2d..5e0218273 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticleTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticleTest.java
@@ -17,6 +17,7 @@ package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
 
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.core.testing.CoreObjects;
+import dev.knative.eventing.kafka.broker.dispatcher.MockReactiveKafkaConsumer;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.RecordDispatcherImpl;
 import dev.knative.eventing.kafka.broker.dispatcher.main.ConsumerVerticleContext;
 import dev.knative.eventing.kafka.broker.dispatcher.main.FakeConsumerVerticleContext;
@@ -25,12 +26,10 @@ import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.junit5.VertxExtension;
-import io.vertx.kafka.client.common.TopicPartition;
-import io.vertx.kafka.client.consumer.KafkaConsumer;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.MockConsumer;
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
+import org.apache.kafka.common.TopicPartition;
 import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -102,7 +101,7 @@ public class OrderedConsumerVerticleTest extends AbstractConsumerVerticleTest {
     final Map<TopicPartition, List<Long>> receivedRecords = new HashMap<>(partitions);
     final var recordDispatcher = mock(RecordDispatcherImpl.class);
     when(recordDispatcher.dispatch(any())).then(invocation -> {
-      final KafkaConsumerRecord<String, CloudEvent> record = invocation.getArgument(0);
+      final ConsumerRecord<String, CloudEvent> record = invocation.getArgument(0);
       return recordDispatcherLogicMock(vertx, random, delay, latch, record, receivedRecords);
     });
     when(recordDispatcher.close()).thenReturn(Future.succeededFuture());
@@ -113,7 +112,7 @@ public class OrderedConsumerVerticleTest extends AbstractConsumerVerticleTest {
         CoreObjects.egress1()
       ),
       (vx, consumerVerticle) -> {
-        consumerVerticle.setConsumer(KafkaConsumer.create(vx, consumer));
+        consumerVerticle.setConsumer(new MockReactiveKafkaConsumer<>(consumer));
         consumerVerticle.setRecordDispatcher(recordDispatcher);
         consumerVerticle.setCloser(Future::succeededFuture);
 
@@ -134,12 +133,12 @@ public class OrderedConsumerVerticleTest extends AbstractConsumerVerticleTest {
     // Assign partitions to consumer (required to add records)
     consumer.updateEndOffsets(
       IntStream.range(0, partitions)
-        .mapToObj(partition -> new org.apache.kafka.common.TopicPartition(topic, partition))
+        .mapToObj(partition -> new TopicPartition(topic, partition))
         .collect(Collectors.toMap(Function.identity(), v -> 0L))
     );
     consumer.rebalance(
       IntStream.range(0, partitions)
-        .mapToObj(partition -> new org.apache.kafka.common.TopicPartition(topic, partition))
+        .mapToObj(partition -> new TopicPartition(topic, partition))
         .collect(Collectors.toList())
     );
 
@@ -202,7 +201,7 @@ public class OrderedConsumerVerticleTest extends AbstractConsumerVerticleTest {
     Random random,
     long millis,
     CountDownLatch latch,
-    KafkaConsumerRecord<String, CloudEvent> record,
+    ConsumerRecord<String, CloudEvent> record,
     Map<TopicPartition, List<Long>> receivedRecords) {
     if (millis == 0) {
       receivedRecords
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 720e4ca0d..e9f04b5a0 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
@@ -17,7 +17,6 @@ package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
 
 import dev.knative.eventing.kafka.broker.dispatcher.main.ConsumerVerticleContext;
 
-import java.util.Set;
 
 public class UnorderedConsumerVerticleTest extends AbstractConsumerVerticleTest {
 
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/integration/ConsumerVerticleFactoryImplMock.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/integration/ConsumerVerticleFactoryImplMock.java
index 33481fc94..75caae85e 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/integration/ConsumerVerticleFactoryImplMock.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/integration/ConsumerVerticleFactoryImplMock.java
@@ -16,26 +16,18 @@
 package dev.knative.eventing.kafka.broker.dispatcher.integration;
 
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
-import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
-import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
 import dev.knative.eventing.kafka.broker.dispatcher.ConsumerVerticleFactory;
-import dev.knative.eventing.kafka.broker.dispatcher.main.ConsumerFactory;
+import dev.knative.eventing.kafka.broker.dispatcher.MockReactiveKafkaConsumer;
+import dev.knative.eventing.kafka.broker.dispatcher.ReactiveKafkaConsumer;
 import dev.knative.eventing.kafka.broker.dispatcher.main.ConsumerVerticleBuilder;
-import dev.knative.eventing.kafka.broker.dispatcher.main.ConsumerVerticleContext;
-import dev.knative.eventing.kafka.broker.dispatcher.main.ConsumerVerticleFactoryImpl;
 import dev.knative.eventing.kafka.broker.dispatcher.main.FakeConsumerVerticleContext;
-import dev.knative.eventing.kafka.broker.dispatcher.main.ProducerFactory;
 import io.cloudevents.CloudEvent;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.Vertx;
-import io.vertx.ext.web.client.WebClientOptions;
-import io.vertx.kafka.client.consumer.KafkaConsumer;
 import io.vertx.kafka.client.producer.KafkaProducer;
 
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Properties;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -47,8 +39,6 @@ import org.apache.kafka.clients.producer.MockProducer;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.serialization.StringSerializer;
 
-import static org.mockito.Mockito.mock;
-
 public class ConsumerVerticleFactoryImplMock implements ConsumerVerticleFactory {
 
   // trigger.id() -> Mock*er
@@ -71,7 +61,12 @@ public class ConsumerVerticleFactoryImplMock implements ConsumerVerticleFactory
     ));
   }
 
-  private KafkaConsumer<Object, CloudEvent> createConsumer(Vertx vertx,
+  /**
+   * @param vertx
+   * @param consumerConfigs
+   * @return
+   */
+  private ReactiveKafkaConsumer<Object, CloudEvent> createConsumer(Vertx vertx,
                                                            Map<String, Object> consumerConfigs) {
     final var consumer = new MockConsumer<Object, CloudEvent>(OffsetResetStrategy.LATEST);
 
@@ -101,7 +96,7 @@ public class ConsumerVerticleFactoryImplMock implements ConsumerVerticleFactory
       );
     });
 
-    return KafkaConsumer.create(vertx, consumer);
+    return new MockReactiveKafkaConsumer<Object, CloudEvent>(consumer);
   }
 
   public void setRecords(final List<ConsumerRecord<String, CloudEvent>> records) {
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImplTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImplTest.java
index a39232271..a9bcb9ece 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImplTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImplTest.java
@@ -20,6 +20,7 @@ import dev.knative.eventing.kafka.broker.contract.DataPlaneContract.BackoffPolic
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressConfig;
 import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
+import dev.knative.eventing.kafka.broker.dispatcher.MockReactiveConsumerFactory;
 import io.cloudevents.kafka.CloudEventDeserializer;
 import io.cloudevents.kafka.CloudEventSerializer;
 import io.cloudevents.kafka.PartitionKeyExtensionInterceptor;
@@ -70,7 +71,8 @@ public class ConsumerVerticleFactoryImplTest {
       new WebClientOptions(),
       producerConfigs,
       mock(AuthProvider.class),
-      mock(MeterRegistry.class)
+      mock(MeterRegistry.class),
+      new MockReactiveConsumerFactory<>()
     );
 
     final var egress = DataPlaneContract.Egress.newBuilder()
@@ -115,7 +117,8 @@ public class ConsumerVerticleFactoryImplTest {
       new WebClientOptions(),
       producerConfigs,
       mock(AuthProvider.class),
-      mock(MeterRegistry.class)
+      mock(MeterRegistry.class),
+      new MockReactiveConsumerFactory<>()
     );
 
     final var egress = DataPlaneContract.Egress.newBuilder()
@@ -164,7 +167,8 @@ public class ConsumerVerticleFactoryImplTest {
       new WebClientOptions(),
       producerConfigs,
       mock(AuthProvider.class),
-      mock(MeterRegistry.class)
+      mock(MeterRegistry.class),
+      new MockReactiveConsumerFactory<>()
     );
 
     final var egress = DataPlaneContract.Egress.newBuilder()
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/FakeConsumerVerticleContext.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/FakeConsumerVerticleContext.java
index ada9bcae9..0c603005c 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/FakeConsumerVerticleContext.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/FakeConsumerVerticleContext.java
@@ -23,6 +23,8 @@ import io.vertx.ext.web.client.WebClientOptions;
 
 import java.util.HashMap;
 
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+
 public class FakeConsumerVerticleContext {
 
   public static ConsumerVerticleContext get() {
@@ -30,7 +32,13 @@ public class FakeConsumerVerticleContext {
       .withProducerConfigs(new HashMap<>())
       .withConsumerConfigs(new HashMap<>())
       .withMeterRegistry(Metrics.getRegistry())
-      .withResource(CoreObjects.resource1(), CoreObjects.egress1());
+      .withResource(CoreObjects.resource1(), CoreObjects.egress1())
+      .withConsumerRebalanceListener(new ConsumerRebalanceListener() {
+        @Override
+        public void onPartitionsRevoked(final java.util.Collection<org.apache.kafka.common.TopicPartition> partitions) {}
+        @Override
+        public void onPartitionsAssigned(final java.util.Collection<org.apache.kafka.common.TopicPartition> partitions) {}
+      });
   }
 
   public static ConsumerVerticleContext get(final DataPlaneContract.Resource resource, final DataPlaneContract.Egress egress) {
@@ -40,6 +48,12 @@ public class FakeConsumerVerticleContext {
       .withAuthProvider(AuthProvider.noAuth())
       .withWebClientOptions(new WebClientOptions())
       .withMeterRegistry(Metrics.getRegistry())
-      .withResource(resource, egress);
+      .withResource(resource, egress)
+      .withConsumerRebalanceListener(new ConsumerRebalanceListener() {
+        @Override
+        public void onPartitionsRevoked(final java.util.Collection<org.apache.kafka.common.TopicPartition> partitions) {}
+        @Override
+        public void onPartitionsAssigned(final java.util.Collection<org.apache.kafka.common.TopicPartition> partitions) {}
+      });
   }
 }
diff --git a/data-plane/tests/pom.xml b/data-plane/tests/pom.xml
index 9a1546487..ba6a319a4 100644
--- a/data-plane/tests/pom.xml
+++ b/data-plane/tests/pom.xml
@@ -54,6 +54,14 @@
       <version>${project.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>dev.knative.eventing.kafka.broker</groupId>
+      <artifactId>dispatcher</artifactId>
+      <classifier>tests</classifier>
+      <type>test-jar</type>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
 
     <dependency>
       <groupId>io.vertx</groupId>
diff --git a/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java b/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java
index 96d3b0a96..a8e9283c6 100644
--- a/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java
+++ b/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java
@@ -21,6 +21,7 @@ import dev.knative.eventing.kafka.broker.core.eventbus.ContractPublisher;
 import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.core.reconciler.impl.ResourcesReconcilerMessageHandler;
 import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
+import dev.knative.eventing.kafka.broker.dispatcher.MockReactiveConsumerFactory;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.CloudEventDeserializer;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.KeyDeserializer;
 import dev.knative.eventing.kafka.broker.dispatcher.main.ConsumerDeployerVerticle;
@@ -351,7 +352,8 @@ public class DataPlaneTest {
       new WebClientOptions(),
       producerConfigs,
       AuthProvider.noAuth(),
-      Metrics.getRegistry()
+      Metrics.getRegistry(),
+      new MockReactiveConsumerFactory<>()
     );
 
     final var verticle = new ConsumerDeployerVerticle(