Repository: strimzi/strimzi-kafka-bridge
Pull request #225 (2019-05-31 09:19:09)

Junit5 migration
This PR introduce junit  migration from version 4 to 5. I changed some values for checkstyle. However, these values should be changed back after next test refactor.

serrss: @scholzj Sorry, I'm not sure that I understand about what things you are talking about.
serrss: @scholzj oh, you mean 2 plugins POM file
scholzj: Exactly.
ppatierno: @Frawless I don't see any further work on this after my comments. Any news?
Frawless: @ppatierno  finishing this one https://github.com/strimzi/strimzi-kafka-operator/pull/1613, after that I will get back to this PR. 

Patch:
diff --git a/pom.xml b/pom.xml
index b6ada85..e9bdb12 100644
--- a/pom.xml
+++ b/pom.xml
@@ -14,10 +14,12 @@
 		<maven.compiler.target>1.8</maven.compiler.target>
 		<slf4j-log4j12.version>1.7.18</slf4j-log4j12.version>
 		<log4j.version>1.2.17</log4j.version>
-		<junit.version>4.12</junit.version>
 		<vertx.version>3.7.1</vertx.version>
 		<debezium.version>0.8.3.Final</debezium.version>
 		<maven.checkstyle.version>3.0.0</maven.checkstyle.version>
+		<jupiter.version>5.4.0</jupiter.version>
+		<junit.platform.version>1.4.0</junit.platform.version>
+		<maven-surefire-plugin.version>2.22.1</maven-surefire-plugin.version>
     </properties>
 
 	<dependencies>
@@ -59,11 +61,30 @@
 
 		<!-- Testing -->
 		<dependency>
-			<groupId>junit</groupId>
-			<artifactId>junit</artifactId>
-			<version>${junit.version}</version>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-api</artifactId>
+			<version>${jupiter.version}</version>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.junit.platform</groupId>
+			<artifactId>junit-platform-runner</artifactId>
+			<version>${junit.platform.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-engine</artifactId>
+			<version>${jupiter.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.junit.jupiter</groupId>
+			<artifactId>junit-jupiter-params</artifactId>
+			<version>${jupiter.version}</version>
+			<scope>test</scope>
+		</dependency>
+
 		<dependency>
 			<groupId>io.vertx</groupId>
 			<artifactId>vertx-unit</artifactId>
@@ -76,6 +97,12 @@
 			<version>${vertx.version}</version>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>io.vertx</groupId>
+			<artifactId>vertx-junit5</artifactId>
+			<version>${vertx.version}</version>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>io.debezium</groupId>
 			<artifactId>debezium-core</artifactId>
@@ -160,6 +187,12 @@
 					</execution>
 				</executions>
 			</plugin>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-surefire-plugin</artifactId>
+				<version>${maven-surefire-plugin.version}</version>
+			</plugin>
+
 		</plugins>
 	</build>
 


Test patch:
diff --git a/src/test/java/io/strimzi/kafka/bridge/KafkaClusterTestBase.java b/src/test/java/io/strimzi/kafka/bridge/KafkaClusterTestBase.java
index e60d9f4..e41e8db 100644
--- a/src/test/java/io/strimzi/kafka/bridge/KafkaClusterTestBase.java
+++ b/src/test/java/io/strimzi/kafka/bridge/KafkaClusterTestBase.java
@@ -9,12 +9,10 @@ import java.io.File;
 import java.io.IOException;
 import java.util.Properties;
 
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-
 import io.debezium.kafka.KafkaCluster;
 import io.debezium.util.Testing;
-import io.vertx.ext.unit.TestContext;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
 
 /**
  * Base class for tests providing a Kafka cluster
@@ -46,14 +44,14 @@ public class KafkaClusterTestBase {
         return kafkaCluster;
     }
 
-    @BeforeClass
-    public static void setUp(TestContext context) throws IOException {
+    @BeforeAll
+    public static void setUp() throws IOException {
         kafkaCluster = kafkaCluster().deleteDataPriorToStartup(true).addBrokers(1).startup();
     }
 
 
-    @AfterClass
-    public static void tearDown(TestContext context) {
+    @AfterAll
+    public static void tearDown() {
 
         if (kafkaCluster != null) {
             kafkaCluster.shutdown();
diff --git a/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeTest.java b/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeTest.java
index cf33150..705a810 100644
--- a/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeTest.java
@@ -14,9 +14,9 @@ import io.strimzi.kafka.bridge.config.KafkaConsumerConfig;
 import io.strimzi.kafka.bridge.converter.DefaultDeserializer;
 import io.strimzi.kafka.bridge.converter.MessageConverter;
 import io.vertx.core.Vertx;
-import io.vertx.ext.unit.Async;
-import io.vertx.ext.unit.TestContext;
-import io.vertx.ext.unit.junit.VertxUnitRunner;
+import io.vertx.junit5.Checkpoint;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
 import io.vertx.kafka.client.consumer.KafkaConsumer;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 import io.vertx.kafka.client.consumer.impl.KafkaConsumerRecordImpl;
@@ -42,11 +42,11 @@ import org.apache.qpid.proton.amqp.messaging.Section;
 import org.apache.qpid.proton.amqp.messaging.Source;
 import org.apache.qpid.proton.amqp.transport.ErrorCondition;
 import org.apache.qpid.proton.message.Message;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Ignore;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -56,10 +56,21 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 
-@RunWith(VertxUnitRunner.class)
-public class AmqpBridgeTest extends KafkaClusterTestBase {
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ExtendWith(VertxExtension.class)
+@SuppressWarnings({"checkstyle:ClassFanOutComplexity", "ClassDataAbstractionCoupling"})
+class AmqpBridgeTest extends KafkaClusterTestBase {
 
     private static final Logger log = LoggerFactory.getLogger(AmqpBridgeTest.class);
 
@@ -84,31 +95,30 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
     private AmqpBridgeConfig bridgeConfigProperties;
 
-    @Before
-    public void before(TestContext context) {
-
+    @BeforeEach
+    void before(VertxTestContext context) {
         this.vertx = Vertx.vertx();
 
         this.bridgeConfigProperties = AmqpBridgeConfig.fromMap(config);
         this.bridge = new AmqpBridge(this.bridgeConfigProperties);
 
-        this.vertx.deployVerticle(this.bridge, context.asyncAssertSuccess());
+        vertx.deployVerticle(this.bridge, context.succeeding(id -> context.completeNow()));
     }
 
-    @After
-    public void after(TestContext context) {
-
-        this.vertx.close(context.asyncAssertSuccess());
+    @AfterEach
+    void after(VertxTestContext context) {
+        vertx.close(context.succeeding(arg -> context.completeNow()));
     }
 
     @Test
-    public void sendSimpleMessages(TestContext context) {
+    void sendSimpleMessages(VertxTestContext context) throws InterruptedException {
         String topic = "sendSimpleMessages";
         kafkaCluster.createTopic(topic, 1, 1);
 
         ProtonClient client = ProtonClient.create(this.vertx);
 
-        Async async = context.async();
+        Checkpoint consume = context.checkpoint();
+
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
             if (ar.succeeded()) {
 
@@ -127,33 +137,35 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                 KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config);
                 consumer.handler(record -> {
-                    context.assertEquals(record.value(), body);
+                    context.verify(() -> assertEquals(record.value(), body));
                     log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                             record.topic(), record.partition(), record.offset(), record.key(), record.value());
                     consumer.close();
-                    async.complete();
+                    consume.flag();
                 });
+
                 consumer.subscribe(topic, done -> {
                     if (!done.succeeded()) {
-                        context.fail(done.cause());
+                        context.failNow(ar.cause());
                     }
                 });
 
                 sender.send(ProtonHelper.tag("my_tag"), message, delivery -> {
                     log.info("Message delivered {}", delivery.getRemoteState());
-                    context.assertEquals(Accepted.getInstance(), delivery.getRemoteState());
+                    context.verify(() -> assertEquals(Accepted.getInstance(), delivery.getRemoteState()));
 
                     sender.close();
                     connection.close();
                 });
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
     @Test
-    public void sendSimpleMessageToPartition(TestContext context) {
+    void sendSimpleMessageToPartition(VertxTestContext context) throws InterruptedException {
         String topic = "sendSimpleMessageToPartition";
         kafkaCluster.createTopic(topic, 2, 1);
 
@@ -163,9 +175,10 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
             e.printStackTrace();
         }
 
+        Checkpoint consume = context.checkpoint();
+
         ProtonClient client = ProtonClient.create(this.vertx);
 
-        Async async = context.async();
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
             if (ar.succeeded()) {
 
@@ -184,17 +197,17 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                 KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config);
                 consumer.handler(record -> {
-                    context.assertEquals(record.value(), body);
+                    context.verify(() -> assertEquals(record.value(), body));
                     // checking the right partition which should not be just the first one (0)
-                    context.assertEquals(record.partition(), 1);
+                    context.verify(() -> assertEquals(record.partition(), 1));
                     log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                             record.topic(), record.partition(), record.offset(), record.key(), record.value());
                     consumer.close();
-                    async.complete();
+                    consume.flag();
                 });
                 consumer.subscribe(topic, done -> {
                     if (!done.succeeded()) {
-                        context.fail(done.cause());
+                        context.failNow(done.cause());
                     }
                 });
 
@@ -206,25 +219,26 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                 sender.send(ProtonHelper.tag("my_tag"), message, delivery -> {
                     log.info("Message delivered {}", delivery.getRemoteState());
-                    context.assertEquals(Accepted.getInstance(), delivery.getRemoteState());
+                    context.verify(() -> assertEquals(Accepted.getInstance(), delivery.getRemoteState()));
 
                     sender.close();
                     connection.close();
                 });
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
     @Test
-    public void sendSimpleMessageWithKey(TestContext context) {
+    void sendSimpleMessageWithKey(VertxTestContext context) throws InterruptedException {
         String topic = "sendSimpleMessageWithKey";
         kafkaCluster.createTopic(topic, 1, 1);
 
         ProtonClient client = ProtonClient.create(this.vertx);
 
-        Async async = context.async();
+        Checkpoint consume = context.checkpoint();
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
             if (ar.succeeded()) {
 
@@ -243,16 +257,16 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                 KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config);
                 consumer.handler(record -> {
-                    context.assertEquals(record.value(), body);
-                    context.assertEquals(record.key(), "my_key");
+                    context.verify(() -> assertEquals(record.value(), body));
+                    context.verify(() -> assertEquals(record.key(), "my_key"));
                     log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                             record.topic(), record.partition(), record.offset(), record.key(), record.value());
                     consumer.close();
-                    async.complete();
+                    consume.flag();
                 });
                 consumer.subscribe(topic, done -> {
                     if (!done.succeeded()) {
-                        context.fail(done.cause());
+                        context.failNow(done.cause());
                     }
                 });
 
@@ -264,25 +278,26 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                 sender.send(ProtonHelper.tag("my_tag"), message, delivery -> {
                     log.info("Message delivered {}", delivery.getRemoteState());
-                    context.assertEquals(Accepted.getInstance(), delivery.getRemoteState());
+                    context.verify(() -> assertEquals(Accepted.getInstance(), delivery.getRemoteState()));
 
                     sender.close();
                     connection.close();
                 });
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
     @Test
-    public void sendBinaryMessage(TestContext context) {
+    void sendBinaryMessage(VertxTestContext context) throws InterruptedException {
         String topic = "sendBinaryMessage";
         kafkaCluster.createTopic(topic, 1, 1);
 
         ProtonClient client = ProtonClient.create(this.vertx);
 
-        Async async = context.async();
+        Checkpoint consume = context.checkpoint();
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
             if (ar.succeeded()) {
 
@@ -300,13 +315,13 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                 KafkaConsumer<String, byte[]> consumer = KafkaConsumer.create(this.vertx, config);
                 consumer.handler(record -> {
-                    context.assertTrue(Arrays.equals(record.value(), value.getBytes()));
+                    context.verify(() -> assertArrayEquals(record.value(), value.getBytes()));
                     consumer.close();
-                    async.complete();
+                    consume.flag();
                 });
                 consumer.subscribe(topic, done -> {
                     if (!done.succeeded()) {
-                        context.fail(done.cause());
+                        context.failNow(done.cause());
                     }
                 });
 
@@ -316,25 +331,26 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                 sender.send(ProtonHelper.tag("my_tag"), message, delivery -> {
                     log.info("Message delivered {}", delivery.getRemoteState());
-                    context.assertEquals(Accepted.getInstance(), delivery.getRemoteState());
+                    context.verify(() -> assertEquals(Accepted.getInstance(), delivery.getRemoteState()));
 
                     sender.close();
                     connection.close();
                 });
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
     @Test
-    public void sendArrayMessage(TestContext context) {
+    void sendArrayMessage(VertxTestContext context) throws InterruptedException {
         String topic = "sendArrayMessage";
         kafkaCluster.createTopic(topic, 1, 1);
 
         ProtonClient client = ProtonClient.create(this.vertx);
 
-        Async async = context.async();
+        Checkpoint consume = context.checkpoint();
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
             if (ar.succeeded()) {
 
@@ -355,13 +371,13 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
                 consumer.handler(record -> {
                     log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                             record.topic(), record.partition(), record.offset(), record.key(), record.value());
-                    context.assertTrue(Arrays.equals(record.value(), array));
+                    context.verify(() -> assertTrue(Arrays.equals(record.value(), array)));
                     consumer.close();
-                    async.complete();
+                    consume.flag();
                 });
                 consumer.subscribe(topic, done -> {
                     if (!done.succeeded()) {
-                        context.fail(done.cause());
+                        context.failNow(done.cause());
                     }
                 });
 
@@ -371,22 +387,23 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                 sender.send(ProtonHelper.tag("my_tag"), message, delivery -> {
                     log.info("Message delivered {}", delivery.getRemoteState());
-                    context.assertEquals(Accepted.getInstance(), delivery.getRemoteState());
+                    context.verify(() -> assertEquals(Accepted.getInstance(), delivery.getRemoteState()));
                 });
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
     @Test
-    public void sendListMessage(TestContext context) {
+    void sendListMessage(VertxTestContext context) throws InterruptedException {
         String topic = "sendListMessage";
         kafkaCluster.createTopic(topic, 1, 1);
 
         ProtonClient client = ProtonClient.create(this.vertx);
 
-        Async async = context.async();
+        Checkpoint consume = context.checkpoint();
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
             if (ar.succeeded()) {
 
@@ -409,13 +426,13 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
                 consumer.handler(record -> {
                     log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                             record.topic(), record.partition(), record.offset(), record.key(), record.value());
-                    context.assertTrue(record.value().equals(list));
+                    context.verify(() -> assertTrue(record.value().equals(list)));
                     consumer.close();
-                    async.complete();
+                    consume.flag();
                 });
                 consumer.subscribe(topic, done -> {
                     if (!done.succeeded()) {
-                        context.fail(done.cause());
+                        context.failNow(done.cause());
                     }
                 });
 
@@ -425,22 +442,23 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                 sender.send(ProtonHelper.tag("my_tag"), message, delivery -> {
                     log.info("Message delivered {}", delivery.getRemoteState());
-                    context.assertEquals(Accepted.getInstance(), delivery.getRemoteState());
+                    context.verify(() -> assertEquals(Accepted.getInstance(), delivery.getRemoteState()));
                 });
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
     @Test
-    public void sendMapMessage(TestContext context) {
+    void sendMapMessage(VertxTestContext context) throws InterruptedException {
         String topic = "sendMapMessage";
         kafkaCluster.createTopic(topic, 1, 1);
 
         ProtonClient client = ProtonClient.create(this.vertx);
 
-        Async async = context.async();
+        Checkpoint consume = context.checkpoint();
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
             if (ar.succeeded()) {
 
@@ -463,13 +481,13 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
                 consumer.handler(record -> {
                     log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                             record.topic(), record.partition(), record.offset(), record.key(), record.value());
-                    context.assertTrue(record.value().equals(map));
+                    context.verify(() -> assertEquals(record.value(), map));
                     consumer.close();
-                    async.complete();
+                    consume.flag();
                 });
                 consumer.subscribe(topic, done -> {
                     if (!done.succeeded()) {
-                        context.fail(done.cause());
+                        context.failNow(done.cause());
                     }
                 });
 
@@ -479,22 +497,23 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                 sender.send(ProtonHelper.tag("my_tag"), message, delivery -> {
                     log.info("Message delivered {}", delivery.getRemoteState());
-                    context.assertEquals(Accepted.getInstance(), delivery.getRemoteState());
+                    context.verify(() -> assertEquals(Accepted.getInstance(), delivery.getRemoteState()));
                 });
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
     @Test
-    public void sendPeriodicMessage(TestContext context) {
+    void sendPeriodicMessage(VertxTestContext context) throws InterruptedException {
         String topic = "sendPeriodicMessage";
         kafkaCluster.createTopic(topic, 1, 1);
 
         ProtonClient client = ProtonClient.create(this.vertx);
 
-        Async async = context.async();
+        Checkpoint consume = context.checkpoint();
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
             if (ar.succeeded()) {
 
@@ -511,18 +530,22 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
                 KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config);
                 consumer.batchHandler(records -> {
 
-                    context.assertEquals(this.count, records.size());
-                    for (int i = 0; i < records.size(); i++) {
-                        KafkaConsumerRecord<String, String> record = records.recordAt(i);
-                        log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
-                                record.topic(), record.partition(), record.offset(), record.key(), record.value());
-                        context.assertEquals("key-" + i, record.key());
-                    }
+                    context.verify(() -> {
+                        assertEquals(this.count, records.size());
+                        for (int i = 0; i < records.size(); i++) {
+                            KafkaConsumerRecord<String, String> record = records.recordAt(i);
+                            log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
+                                    record.topic(), record.partition(), record.offset(), record.key(), record.value());
+                            int finalI = i;
+                            assertEquals("key-" + finalI, record.key());
+                        }
+                    });
 
                     consumer.close();
-                    async.complete();
+                    consume.flag();
+                });
+                consumer.handler(record -> {
                 });
-                consumer.handler(record -> { });
 
                 this.count = 0;
 
@@ -531,7 +554,7 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
                     if (connection.isDisconnected()) {
                         this.vertx.cancelTimer(timerId);
                         // test failed
-                        context.assertTrue(false);
+                        context.verify(() -> assertTrue(false));
                     } else {
 
                         if (this.count < AmqpBridgeTest.PERIODIC_MAX_MESSAGE) {
@@ -544,9 +567,9 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
                             Message message = ProtonHelper.message(topic, "Periodic message [" + this.count + "] from " + connection.getContainer());
                             message.setMessageAnnotations(messageAnnotations);
 
-                            sender.send(ProtonHelper.tag("my_tag_" + String.valueOf(this.count)), message, delivery -> {
+                            sender.send(ProtonHelper.tag("my_tag_" + this.count), message, delivery -> {
                                 log.info("Message delivered {}", delivery.getRemoteState());
-                                context.assertEquals(Accepted.getInstance(), delivery.getRemoteState());
+                                context.verify(() -> assertEquals(Accepted.getInstance(), delivery.getRemoteState()));
                             });
 
                             this.count++;
@@ -557,26 +580,27 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
                             // subscribe Kafka consumer for getting messages
                             consumer.subscribe(topic, done -> {
                                 if (!done.succeeded()) {
-                                    context.fail(done.cause());
+                                    context.failNow(done.cause());
                                 }
                             });
                         }
                     }
                 });
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
     @Test
-    public void sendReceiveInMultiplexing(TestContext context) {
+    void sendReceiveInMultiplexing(VertxTestContext context) throws InterruptedException {
         String topic = "sendReceiveInMultiplexing";
         kafkaCluster.createTopic(topic, 1, 1);
 
         ProtonClient client = ProtonClient.create(this.vertx);
 
-        Async async = context.async();
+        Checkpoint consume = context.checkpoint();
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
 
             if (ar.succeeded()) {
@@ -596,40 +620,43 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
                         log.info("Message received {}", new String(value));
                         // default is AT_LEAST_ONCE QoS (unsettled) so we need to send disposition (settle) to sender
                         delivery.disposition(Accepted.getInstance(), true);
-                        context.assertEquals(sentBody, new String(value));
-                        async.complete();
+                        context.verify(() -> assertEquals(sentBody, new String(value)));
+                        consume.flag();
                     }
                 })
-                .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
+                        .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
 
                 ProtonSender sender = connection.createSender(null);
                 sender.open();
 
                 sender.send(ProtonHelper.tag("my_tag"), sentMessage, delivery -> {
                     log.info("Message delivered {}", delivery.getRemoteState());
-                    context.assertEquals(Accepted.getInstance(), delivery.getRemoteState());
+                    context.verify(() -> assertEquals(Accepted.getInstance(), delivery.getRemoteState()));
                 });
 
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
     @Test
-    public void receiveSimpleMessage(TestContext context) {
+    void receiveSimpleMessage(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "receiveSimpleMessage";
         kafkaCluster.createTopic(topic, 1, 1);
 
         String sentBody = "Simple message";
 
-        Async send = context.async();
-        kafkaCluster.useTo().produceStrings(1, send::complete, () ->
+        // Futures for wait
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
+        Checkpoint consume = context.checkpoint();
+
+        kafkaCluster.useTo().produceStrings(1, () -> produce.complete(true), () ->
                 new ProducerRecord<>(topic, 0, null, sentBody));
-        send.await();
+        produce.get(60, TimeUnit.SECONDS);
 
         ProtonClient client = ProtonClient.create(this.vertx);
-        Async async = context.async();
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
             if (ar.succeeded()) {
 
@@ -648,47 +675,52 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                         // get topic, partition, offset and key from AMQP annotations
                         MessageAnnotations annotations = message.getMessageAnnotations();
-                        context.assertNotNull(annotations);
-                        String topicAnnotation = String.valueOf(annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_TOPIC_ANNOTATION)));
-                        context.assertNotNull(topicAnnotation);
-                        Integer partitionAnnotation = (Integer) annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_PARTITION_ANNOTATION));
-                        context.assertNotNull(partitionAnnotation);
-                        Long offsetAnnotation = (Long) annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_OFFSET_ANNOTATION));
-                        context.assertNotNull(offsetAnnotation);
-                        Object keyAnnotation = annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_KEY_ANNOTATION));
-                        context.assertNull(keyAnnotation);
-                        log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
-                                topicAnnotation, partitionAnnotation, offsetAnnotation, keyAnnotation, new String(value));
-
-                        context.assertEquals(topicAnnotation, topic);
-                        context.assertEquals(partitionAnnotation, 0);
-                        context.assertEquals(offsetAnnotation, 0L);
-                        context.assertEquals(sentBody, new String(value));
-                        async.complete();
+                        context.verify(() -> {
+                            assertNotNull(annotations);
+                            String topicAnnotation = String.valueOf(annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_TOPIC_ANNOTATION)));
+                            assertNotNull(topicAnnotation);
+                            Integer partitionAnnotation = (Integer) annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_PARTITION_ANNOTATION));
+                            assertNotNull(partitionAnnotation);
+                            Long offsetAnnotation = (Long) annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_OFFSET_ANNOTATION));
+                            assertNotNull(offsetAnnotation);
+                            Object keyAnnotation = annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_KEY_ANNOTATION));
+                            assertNull(keyAnnotation);
+                            log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
+                                    topicAnnotation, partitionAnnotation, offsetAnnotation, keyAnnotation, new String(value));
+
+                            assertEquals(topicAnnotation, topic);
+                            assertEquals(partitionAnnotation, 0);
+                            assertEquals(offsetAnnotation, 0L);
+                            assertEquals(sentBody, new String(value));
+                        });
+                        consume.flag();
                     }
                 })
-                .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
+                        .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
     @Test
-    public void receiveSimpleMessageFromPartition(TestContext context) {
+    void receiveSimpleMessageFromPartition(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "receiveSimpleMessageFromPartition";
         kafkaCluster.createTopic(topic, 2, 1);
 
         String sentBody = "Simple message";
 
-        Async send = context.async();
-        kafkaCluster.useTo().produceStrings(1, send::complete, () ->
+        // Futures for wait
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
+        Checkpoint consume = context.checkpoint();
+
+        kafkaCluster.useTo().produceStrings(1, () -> produce.complete(true), () ->
                 new ProducerRecord<>(topic, 1, null, sentBody));
-        send.await();
+        produce.get(60, TimeUnit.SECONDS);
 
         ProtonClient client = ProtonClient.create(this.vertx);
 
-        Async async = context.async();
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
             if (ar.succeeded()) {
 
@@ -715,46 +747,48 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                         // get topic, partition, offset and key from AMQP annotations
                         MessageAnnotations annotations = message.getMessageAnnotations();
-                        context.assertNotNull(annotations);
+                        context.verify(() -> assertNotNull(annotations));
                         String topicAnnotation = String.valueOf(annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_TOPIC_ANNOTATION)));
-                        context.assertNotNull(topicAnnotation);
+                        assertNotNull(topicAnnotation);
                         Integer partitionAnnotation = (Integer) annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_PARTITION_ANNOTATION));
-                        context.assertNotNull(partitionAnnotation);
+                        assertNotNull(partitionAnnotation);
                         Long offsetAnnotation = (Long) annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_OFFSET_ANNOTATION));
-                        context.assertNotNull(offsetAnnotation);
+                        assertNotNull(offsetAnnotation);
                         Object keyAnnotation = annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_KEY_ANNOTATION));
-                        context.assertNull(keyAnnotation);
+                        assertNull(keyAnnotation);
                         log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                                 topicAnnotation, partitionAnnotation, offsetAnnotation, keyAnnotation, new String(value));
 
-                        context.assertEquals(topicAnnotation, topic);
-                        context.assertEquals(partitionAnnotation, 1);
-                        context.assertEquals(offsetAnnotation, 0L);
-                        context.assertEquals(sentBody, new String(value));
-                        async.complete();
+                        assertEquals(topicAnnotation, topic);
+                        assertEquals(partitionAnnotation, 1);
+                        assertEquals(offsetAnnotation, 0L);
+                        assertEquals(sentBody, new String(value));
+                        consume.flag();
                     }
                 })
-                .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
+                        .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
     @Test
-    public void receiveSimpleMessageFromPartitionAndOffset(TestContext context) {
+    void receiveSimpleMessageFromPartitionAndOffset(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "receiveSimpleMessageFromPartitionAndOffset";
         kafkaCluster.createTopic(topic, 1, 1);
 
-        Async batch = context.async();
+        // Futures for wait
+        CompletableFuture<Boolean> batch = new CompletableFuture<>();
+        Checkpoint consume = context.checkpoint();
         AtomicInteger index = new AtomicInteger();
-        kafkaCluster.useTo().produceStrings(11, batch::complete,  () ->
+        kafkaCluster.useTo().produceStrings(11, () -> batch.complete(true), () ->
                 new ProducerRecord<>(topic, 0, "key-" + index.get(), "value-" + index.getAndIncrement()));
-        batch.awaitSuccess(10000);
+        batch.get(60, TimeUnit.SECONDS);
 
         ProtonClient client = ProtonClient.create(this.vertx);
 
-        Async async = context.async();
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
             if (ar.succeeded()) {
 
@@ -782,41 +816,42 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 
                         // get topic, partition, offset and key from AMQP annotations
                         MessageAnnotations annotations = message.getMessageAnnotations();
-                        context.assertNotNull(annotations);
+                        context.verify(() -> assertNotNull(annotations));
                         String topicAnnotation = String.valueOf(annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_TOPIC_ANNOTATION)));
-                        context.assertNotNull(topicAnnotation);
+                        assertNotNull(topicAnnotation);
                         Integer partitionAnnotation = (Integer) annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_PARTITION_ANNOTATION));
-                        context.assertNotNull(partitionAnnotation);
+                        assertNotNull(partitionAnnotation);
                         Long offsetAnnotation = (Long) annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_OFFSET_ANNOTATION));
-                        context.assertNotNull(offsetAnnotation);
+                        assertNotNull(offsetAnnotation);
                         Object keyAnnotation = annotations.getValue().get(Symbol.valueOf(AmqpBridge.AMQP_KEY_ANNOTATION));
-                        context.assertNotNull(keyAnnotation);
+                        assertNotNull(keyAnnotation);
                         log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                                 topicAnnotation, partitionAnnotation, offsetAnnotation, keyAnnotation, new String(value));
 
-                        context.assertEquals(topicAnnotation, topic);
-                        context.assertEquals(partitionAnnotation, 0);
-                        context.assertEquals(offsetAnnotation, 10L);
-                        context.assertEquals(keyAnnotation, "key-10");
-                        context.assertEquals("value-10", new String(value));
-                        async.complete();
+                        assertEquals(topicAnnotation, topic);
+                        assertEquals(partitionAnnotation, 0);
+                        assertEquals(offsetAnnotation, 10L);
+                        assertEquals(keyAnnotation, "key-10");
+                        assertEquals("value-10", new String(value));
+                        consume.flag();
                     }
                 })
-                .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
+                        .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
+    @Disabled
     @Test
-    @Ignore
-    public void noPartitionsAvailable(TestContext context) {
+    void noPartitionsAvailable(VertxTestContext context) throws InterruptedException {
         String topic = "noPartitionsAvailable";
         kafkaCluster.createTopic(topic, 1, 1);
 
         ProtonClient client = ProtonClient.create(this.vertx);
-        Async async = context.async();
+        Checkpoint noPartition = context.checkpoint();
         client.connect(AmqpBridgeTest.BRIDGE_HOST, AmqpBridgeTest.BRIDGE_PORT, ar -> {
             if (ar.succeeded()) {
 
@@ -832,41 +867,45 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
                     ProtonReceiver receiver1 = connection.createReceiver(topic + "/group.id/my_group");
                     receiver1.closeHandler(ar1 -> {
                         if (ar1.succeeded()) {
-                            context.fail(ar1.cause());
+                            context.failNow(ar1.cause());
                         } else {
                             ErrorCondition condition = receiver1.getRemoteCondition();
                             log.info(condition.getDescription());
-                            context.assertEquals(condition.getCondition(), Symbol.getSymbol(AmqpBridge.AMQP_ERROR_NO_PARTITIONS));
-                            async.complete();
+                            context.verify(() -> assertEquals(condition.getCondition(), Symbol.getSymbol(AmqpBridge.AMQP_ERROR_NO_PARTITIONS)));
+                            noPartition.flag();
                         }
                     })
-                    .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
+                            .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
                 });
 
 
             } else {
-                context.fail(ar.cause());
+                context.failNow(ar.cause());
             }
         });
+        assertTrue(context.awaitCompletion(60, TimeUnit.SECONDS));
     }
 
     @Test
-    public void defaultMessageConverterNullKeyTest(TestContext context) {
+    void defaultMessageConverterNullKeyTest(VertxTestContext context) {
         MessageConverter defaultMessageConverter = new AmqpDefaultMessageConverter();
-        context.assertNull(convertedMessageWithNullKey(defaultMessageConverter));
+        context.verify(() -> assertNull(convertedMessageWithNullKey(defaultMessageConverter)));
+        context.completeNow();
     }
 
     @Test
-    public void jsonMessageConverterNullKeyTest(TestContext context) {
+    void jsonMessageConverterNullKeyTest(VertxTestContext context) {
         MessageConverter jsonMessageConverter = new AmqpJsonMessageConverter();
-        context.assertNull(convertedMessageWithNullKey(jsonMessageConverter));
+        context.verify(() -> assertNull(convertedMessageWithNullKey(jsonMessageConverter)));
+        context.completeNow();
     }
 
-    @Ignore
+    @Disabled
     @Test
-    public void rawMessageConverterNullKeyTest(TestContext context) {
+    void rawMessageConverterNullKeyTest(VertxTestContext context) {
         MessageConverter rawMessageConverter = new AmqpRawMessageConverter();
-        context.assertNull(convertedMessageWithNullKey(rawMessageConverter));
+        context.verify(() -> assertNull(convertedMessageWithNullKey(rawMessageConverter)));
+        context.completeNow();
     }
 
     private Object convertedMessageWithNullKey(MessageConverter messageConverter) {
diff --git a/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java b/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java
index c040794..289baac 100644
--- a/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java
@@ -34,8 +34,8 @@ import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
 import org.apache.qpid.proton.amqp.messaging.Source;
 import org.apache.qpid.proton.amqp.transport.ErrorCondition;
 import org.apache.qpid.proton.message.Message;
-import org.junit.Ignore;
-import org.junit.Test;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 
 import java.lang.reflect.Field;
@@ -56,7 +56,7 @@ import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-public class AmqpSinkBridgeEndpointMockTest {
+class AmqpSinkBridgeEndpointMockTest {
 
     private static Map<String, Object> config = new HashMap<>();
 
@@ -151,7 +151,7 @@ public class AmqpSinkBridgeEndpointMockTest {
 
     // Test normal flow in AT_MOST_ONCE mode.
     @Test
-    @Ignore
+    @Disabled
     public <K, V> void normalFlow_AtMostOnce() throws Exception {
         String topic = "my_topic";
         Vertx vertx = Vertx.vertx();
@@ -572,7 +572,7 @@ public class AmqpSinkBridgeEndpointMockTest {
     // What happens if we can't get the partitions for the given topic?
     // @throws AmqpErrorConditionException
     @Test
-    @Ignore
+    @Disabled
     public <K, V> void partitionsForFails() throws Exception {
         String topic = "my_topic";
         Vertx vertx = Vertx.vertx();
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
index 122a1db..ad378e9 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
@@ -16,12 +16,11 @@ import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
-import io.vertx.ext.unit.Async;
-import io.vertx.ext.unit.TestContext;
-import io.vertx.ext.unit.junit.VertxUnitRunner;
 import io.vertx.ext.web.client.HttpResponse;
 import io.vertx.ext.web.client.WebClient;
 import io.vertx.ext.web.codec.BodyCodec;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
 import io.vertx.kafka.client.consumer.KafkaConsumer;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -30,25 +29,36 @@ import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Ignore;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.junit.jupiter.api.extension.ExtendWith;
+
 import javax.xml.bind.DatatypeConverter;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
-@RunWith(VertxUnitRunner.class)
-@SuppressWarnings("checkstyle:JavaNCSS")
-public class HttpBridgeTest extends KafkaClusterTestBase {
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ExtendWith(VertxExtension.class)
+@SuppressWarnings({"checkstyle:JavaNCSS"})
+class HttpBridgeTest extends KafkaClusterTestBase {
 
     private static final Logger log = LoggerFactory.getLogger(HttpBridgeTest.class);
 
@@ -67,6 +77,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
     private static final int PERIODIC_MAX_MESSAGE = 10;
     private static final int PERIODIC_DELAY = 200;
     private static final int MULTIPLE_MAX_MESSAGE = 10;
+    private static final int TEST_TIMEOUT = 60;
     private int count;
 
     private Vertx vertx;
@@ -74,31 +85,25 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
     private HttpBridgeConfig bridgeConfigProperties;
 
-    @Before
-    public void before(TestContext context) {
-
-        vertx = Vertx.vertx();
-
+    @BeforeEach
+    void before(VertxTestContext context) {
+        this.vertx = Vertx.vertx();
         this.bridgeConfigProperties = HttpBridgeConfig.fromMap(config);
         this.httpBridge = new HttpBridge(this.bridgeConfigProperties);
 
-        this.vertx.deployVerticle(this.httpBridge, context.asyncAssertSuccess());
-
+        vertx.deployVerticle(this.httpBridge, context.succeeding(id -> context.completeNow()));
     }
 
-    @After
-    public void after(TestContext context) {
-
-        this.vertx.close(context.asyncAssertSuccess());
+    @AfterEach
+    void after(VertxTestContext context) {
+        vertx.close(context.succeeding(arg -> context.completeNow()));
     }
 
     @Test
-    public void sendSimpleMessage(TestContext context) {
+    void sendSimpleMessage(VertxTestContext context) throws Throwable {
         String topic = "sendSimpleMessage";
         kafkaCluster.createTopic(topic, 1, 1);
 
-        Async async = context.async();
-
         String value = "message-value";
 
         JsonArray records = new JsonArray();
@@ -116,16 +121,17 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
 
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-
-                    JsonArray offsets = bridgeResponse.getJsonArray("offsets");
-                    context.assertEquals(1, offsets.size());
-                    JsonObject metadata = offsets.getJsonObject(0);
-                    context.assertEquals(0, metadata.getInteger("partition"));
-                    context.assertEquals(0L, metadata.getLong("offset"));
+                        JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                        assertEquals(1, offsets.size());
+                        JsonObject metadata = offsets.getJsonObject(0);
+                        assertEquals(0, metadata.getInteger("partition"));
+                        assertEquals(0L, metadata.getLong("offset"));
+                    });
                 });
 
         Properties config = kafkaCluster.useTo().getConsumerProperties("groupId", null, OffsetResetStrategy.EARLIEST);
@@ -133,31 +139,31 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config,
                 new StringDeserializer(), new KafkaJsonDeserializer<>(String.class));
         consumer.handler(record -> {
-            context.assertEquals(value, record.value());
-            context.assertEquals(topic, record.topic());
-            context.assertEquals(0, record.partition());
-            context.assertEquals(0L, record.offset());
-            context.assertNull(record.key());
+            context.verify(() -> {
+                assertEquals(value, record.value());
+                assertEquals(topic, record.topic());
+                assertEquals(0, record.partition());
+                assertEquals(0L, record.offset());
+                assertNull(record.key());
+            });
             log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                     record.topic(), record.partition(), record.offset(), record.key(), record.value());
             consumer.close();
-            async.complete();
+            context.completeNow();
         });
 
         consumer.subscribe(topic, done -> {
             if (!done.succeeded()) {
-                context.fail(done.cause());
+                context.failNow(done.cause());
             }
         });
     }
 
     @Test
-    public void sendSimpleMessageToPartition(TestContext context) {
+    void sendSimpleMessageToPartition(VertxTestContext context) {
         String topic = "sendSimpleMessageToPartition";
         kafkaCluster.createTopic(topic, 2, 1);
 
-        Async async = context.async();
-
         String value = "message-value";
 
         int partition = 1;
@@ -178,16 +184,18 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
 
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
 
-                    JsonArray offsets = bridgeResponse.getJsonArray("offsets");
-                    context.assertEquals(1, offsets.size());
-                    JsonObject metadata = offsets.getJsonObject(0);
-                    context.assertEquals(partition, metadata.getInteger("partition"));
-                    context.assertEquals(0L, metadata.getLong("offset"));
+                        JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                        assertEquals(1, offsets.size());
+                        JsonObject metadata = offsets.getJsonObject(0);
+                        assertEquals(partition, metadata.getInteger("partition"));
+                        assertEquals(0L, metadata.getLong("offset"));
+                    });
                 });
 
         Properties config = kafkaCluster.useTo().getConsumerProperties("groupId", null, OffsetResetStrategy.EARLIEST);
@@ -195,31 +203,31 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config,
                 new StringDeserializer(), new KafkaJsonDeserializer<>(String.class));
         consumer.handler(record -> {
-            context.assertEquals(value, record.value());
-            context.assertEquals(topic, record.topic());
-            context.assertEquals(partition, record.partition());
-            context.assertEquals(0L, record.offset());
-            context.assertNull(record.key());
+            context.verify(() -> {
+                assertEquals(value, record.value());
+                assertEquals(topic, record.topic());
+                assertEquals(partition, record.partition());
+                assertEquals(0L, record.offset());
+                assertNull(record.key());
+            });
             log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                     record.topic(), record.partition(), record.offset(), record.key(), record.value());
             consumer.close();
-            async.complete();
+            context.completeNow();
         });
 
         consumer.subscribe(topic, done -> {
             if (!done.succeeded()) {
-                context.fail(done.cause());
+                context.failNow(done.cause());
             }
         });
     }
 
     @Test
-    public void sendSimpleMessageWithKey(TestContext context) {
+    void sendSimpleMessageWithKey(VertxTestContext context) {
         String topic = "sendSimpleMessageWithKey";
         kafkaCluster.createTopic(topic, 2, 1);
 
-        Async async = context.async();
-
         String value = "message-value";
         String key = "my-key";
 
@@ -239,16 +247,17 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
 
-                    JsonArray offsets = bridgeResponse.getJsonArray("offsets");
-                    context.assertEquals(1, offsets.size());
-                    JsonObject metadata = offsets.getJsonObject(0);
-                    context.assertNotNull(metadata.getInteger("partition"));
-                    context.assertEquals(0L, metadata.getLong("offset"));
+                        JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                        assertEquals(1, offsets.size());
+                        JsonObject metadata = offsets.getJsonObject(0);
+                        assertNotNull(metadata.getInteger("partition"));
+                        assertEquals(0L, metadata.getLong("offset"));
+                    });
                 });
 
         Properties config = kafkaCluster.useTo().getConsumerProperties("groupId", null, OffsetResetStrategy.EARLIEST);
@@ -256,31 +265,31 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config,
                 new KafkaJsonDeserializer<>(String.class), new KafkaJsonDeserializer<>(String.class));
         consumer.handler(record -> {
-            context.assertEquals(value, record.value());
-            context.assertEquals(topic, record.topic());
-            context.assertNotNull(record.partition());
-            context.assertEquals(0L, record.offset());
-            context.assertEquals(record.key(), key);
+            context.verify(() -> {
+                assertEquals(value, record.value());
+                assertEquals(topic, record.topic());
+                assertNotNull(record.partition());
+                assertEquals(0L, record.offset());
+                assertEquals(record.key(), key);
+            });
             log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                     record.topic(), record.partition(), record.offset(), record.key(), record.value());
             consumer.close();
-            async.complete();
+            context.completeNow();
         });
 
         consumer.subscribe(topic, done -> {
             if (!done.succeeded()) {
-                context.fail(done.cause());
+                context.failNow(done.cause());
             }
         });
     }
 
     @Test
-    public void sendBinaryMessageWithKey(TestContext context) {
+    void sendBinaryMessageWithKey(VertxTestContext context) {
         String topic = "sendBinaryMessageWithKey";
         kafkaCluster.createTopic(topic, 2, 1);
 
-        Async async = context.async();
-
         String value = "message-value";
         String key = "my-key";
 
@@ -300,16 +309,17 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_BINARY)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
 
-                    JsonArray offsets = bridgeResponse.getJsonArray("offsets");
-                    context.assertEquals(1, offsets.size());
-                    JsonObject metadata = offsets.getJsonObject(0);
-                    context.assertNotNull(metadata.getInteger("partition"));
-                    context.assertEquals(0L, metadata.getLong("offset"));
+                        JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                        assertEquals(1, offsets.size());
+                        JsonObject metadata = offsets.getJsonObject(0);
+                        assertNotNull(metadata.getInteger("partition"));
+                        assertEquals(0L, metadata.getLong("offset"));
+                    });
                 });
 
         Properties config = kafkaCluster.useTo().getConsumerProperties("groupId", null, OffsetResetStrategy.EARLIEST);
@@ -317,31 +327,32 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         KafkaConsumer<byte[], byte[]> consumer = KafkaConsumer.create(this.vertx, config,
                 new ByteArrayDeserializer(), new ByteArrayDeserializer());
         consumer.handler(record -> {
-            context.assertEquals(value, new String(record.value()));
-            context.assertEquals(topic, record.topic());
-            context.assertNotNull(record.partition());
-            context.assertEquals(0L, record.offset());
-            context.assertEquals(key, new String(record.key()));
+            context.verify(() -> {
+                assertEquals(value, new String(record.value()));
+                assertEquals(topic, record.topic());
+                assertNotNull(record.partition());
+                assertEquals(0L, record.offset());
+                assertEquals(key, new String(record.key()));
+            });
+
             log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                     record.topic(), record.partition(), record.offset(), record.key(), record.value());
             consumer.close();
-            async.complete();
+            context.completeNow();
         });
 
         consumer.subscribe(topic, done -> {
             if (!done.succeeded()) {
-                context.fail(done.cause());
+                context.failNow(done.cause());
             }
         });
     }
 
     @Test
-    public void sendPeriodicMessage(TestContext context) {
+    void sendPeriodicMessage(VertxTestContext context) {
         String topic = "sendPeriodicMessage";
         kafkaCluster.createTopic(topic, 1, 1);
 
-        Async async = context.async();
-
         WebClient client = WebClient.create(vertx);
 
         Properties config = kafkaCluster.useTo().getConsumerProperties("groupId", null, OffsetResetStrategy.EARLIEST);
@@ -375,39 +386,39 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
                 consumer.subscribe(topic, done -> {
                     if (!done.succeeded()) {
-                        context.fail(done.cause());
+                        context.failNow(done.cause());
                     }
                 });
             }
         });
 
         consumer.batchHandler(records -> {
-            context.assertEquals(this.count, records.size());
-            for (int i = 0; i < records.size(); i++) {
-                KafkaConsumerRecord<String, String> record = records.recordAt(i);
-                log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
-                        record.topic(), record.partition(), record.offset(), record.key(), record.value());
-                context.assertEquals(record.value(), "Periodic message [" + i + "]");
-                context.assertEquals(topic, record.topic());
-                context.assertNotNull(record.partition());
-                context.assertNotNull(record.offset());
-                context.assertEquals(record.key(), "key-" + i);
-            }
+            context.verify(() -> {
+                assertEquals(this.count, records.size());
+                for (int i = 0; i < records.size(); i++) {
+                    KafkaConsumerRecord<String, String> record = records.recordAt(i);
+                    log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
+                            record.topic(), record.partition(), record.offset(), record.key(), record.value());
+                    assertEquals(record.value(), "Periodic message [" + i + "]");
+                    assertEquals(topic, record.topic());
+                    assertNotNull(record.partition());
+                    assertNotNull(record.offset());
+                    assertEquals(record.key(), "key-" + i);
+                }
+            });
 
             consumer.close();
-            async.complete();
+            context.completeNow();
         });
 
         consumer.handler(record -> { });
     }
 
     @Test
-    public void sendMultipleMessages(TestContext context) {
+    void sendMultipleMessages(VertxTestContext context) {
         String topic = "sendMultipleMessages";
         kafkaCluster.createTopic(topic, 1, 1);
 
-        Async async = context.async();
-
         String value = "message-value";
 
         int numMessages = MULTIPLE_MAX_MESSAGE;
@@ -428,17 +439,17 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> assertTrue(ar.succeeded()));
 
                     HttpResponse<JsonObject> response = ar.result();
                     JsonObject bridgeResponse = response.body();
 
                     JsonArray offsets = bridgeResponse.getJsonArray("offsets");
-                    context.assertEquals(numMessages, offsets.size());
+                    assertEquals(numMessages, offsets.size());
                     for (int i = 0; i < numMessages; i++) {
                         JsonObject metadata = offsets.getJsonObject(i);
-                        context.assertEquals(0, metadata.getInteger("partition"));
-                        context.assertNotNull(metadata.getLong("offset"));
+                        assertEquals(0, metadata.getInteger("partition"));
+                        assertNotNull(metadata.getLong("offset"));
                     }
                 });
 
@@ -448,32 +459,32 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 new StringDeserializer(), new KafkaJsonDeserializer<String>(String.class));
         this.count = 0;
         consumer.handler(record -> {
-            context.assertEquals(value + "-" + this.count++, record.value());
-            context.assertEquals(topic, record.topic());
-            context.assertNotNull(record.partition());
-            context.assertNotNull(record.offset());
-            context.assertNull(record.key());
+            context.verify(() -> {
+                assertEquals(value + "-" + this.count++, record.value());
+                assertEquals(topic, record.topic());
+                assertNotNull(record.partition());
+                assertNotNull(record.offset());
+                assertNull(record.key());
+            });
+
             log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                     record.topic(), record.partition(), record.offset(), record.key(), record.value());
 
             if (this.count == numMessages) {
                 consumer.close();
-                async.complete();
+                context.completeNow();
             }
         });
 
         consumer.subscribe(topic, done -> {
             if (!done.succeeded()) {
-                context.fail(done.cause());
+                context.failNow(done.cause());
             }
         });
     }
 
     @Test
-    public void createConsumer(TestContext context) {
-
-        Async creationAsync = context.async();
-
+    void createConsumer(VertxTestContext context) throws InterruptedException {
         WebClient client = WebClient.create(vertx);
 
         String name = "my-kafka-consumer";
@@ -492,33 +503,34 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
 
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.completeNow();
                 });
 
-        creationAsync.await();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
     }
 
     @Test
-    public void receiveSimpleMessage(TestContext context) {
+    void receiveSimpleMessage(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "receiveSimpleMessage";
         kafkaCluster.createTopic(topic, 1, 1);
 
         String sentBody = "Simple message";
 
-        Async send = context.async();
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send::complete, () -> new ProducerRecord<>(topic, 0, null, sentBody));
-        send.await();
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, 0, null, sentBody));
 
-        Async creationAsync = context.async();
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -531,105 +543,106 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("name", name);
         json.put("format", "json");
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
 
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    create.complete(true);
                 });
 
-        creationAsync.await();
-
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonArray topics = new JsonArray();
         topics.add(topic);
 
         JsonObject topicsRoot = new JsonObject();
         topicsRoot.put("topics", topics);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+
+                    subscribe.complete(true);
                 });
 
-        subscriberAsync.await();
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonArray> response = ar.result();
+                        JsonObject jsonResponse = response.body().getJsonObject(0);
 
-                    HttpResponse<JsonArray> response = ar.result();
-                    JsonObject jsonResponse = response.body().getJsonObject(0);
-
-                    String kafkaTopic = jsonResponse.getString("topic");
-                    int kafkaPartition = jsonResponse.getInteger("partition");
-                    String key = jsonResponse.getString("key");
-                    String value = jsonResponse.getString("value");
-                    long offset = jsonResponse.getLong("offset");
-
-                    context.assertEquals(topic, kafkaTopic);
-                    context.assertEquals(sentBody, value);
-                    context.assertEquals(0L, offset);
-                    context.assertNotNull(kafkaPartition);
-                    context.assertNull(key);
+                        String kafkaTopic = jsonResponse.getString("topic");
+                        int kafkaPartition = jsonResponse.getInteger("partition");
+                        String key = jsonResponse.getString("key");
+                        String value = jsonResponse.getString("value");
+                        long offset = jsonResponse.getLong("offset");
 
-                    consumeAsync.complete();
+                        assertEquals(topic, kafkaTopic);
+                        assertEquals(sentBody, value);
+                        assertEquals(0L, offset);
+                        assertNotNull(kafkaPartition);
+                        assertNull(key);
+                    });
+                    consume.complete(true);
                 });
 
-        consumeAsync.await();
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> assertTrue(ar.succeeded()));
 
                     HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
 
-                    deleteAsync.complete();
+                    delete.complete(true);
                 });
-
-        deleteAsync.await();
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
     }
 
     @Test
-    public void receiveBinaryMessage(TestContext context) {
+    void receiveBinaryMessage(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "receiveBinaryMessage";
         kafkaCluster.createTopic(topic, 1, 1);
 
         String sentBody = "Simple message";
 
-        Async send = context.async();
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         kafkaCluster.useTo().produce("", 1, new ByteArraySerializer(), new ByteArraySerializer(),
-                send::complete, () -> new ProducerRecord<>(topic, 0, null, sentBody.getBytes()));
-        send.await();
-
-        Async creationAsync = context.async();
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, 0, null, sentBody.getBytes()));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -642,94 +655,97 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("name", name);
         json.put("format", "binary");
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonArray topics = new JsonArray();
         topics.add(topic);
 
         JsonObject topicsRoot = new JsonObject();
         topicsRoot.put("topics", topics);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
 
-        subscriberAsync.await();
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_BINARY)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonArray> response = ar.result();
-                    JsonObject jsonResponse = response.body().getJsonObject(0);
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonArray> response = ar.result();
+                        JsonObject jsonResponse = response.body().getJsonObject(0);
 
-                    String kafkaTopic = jsonResponse.getString("topic");
-                    int kafkaPartition = jsonResponse.getInteger("partition");
-                    String key = jsonResponse.getString("key");
-                    String value = new String(DatatypeConverter.parseBase64Binary(jsonResponse.getString("value")));
-                    long offset = jsonResponse.getLong("offset");
-
-                    context.assertEquals(topic, kafkaTopic);
-                    context.assertEquals(sentBody, value);
-                    context.assertEquals(0L, offset);
-                    context.assertNotNull(kafkaPartition);
-                    context.assertNull(key);
+                        String kafkaTopic = jsonResponse.getString("topic");
+                        int kafkaPartition = jsonResponse.getInteger("partition");
+                        String key = jsonResponse.getString("key");
+                        String value = new String(DatatypeConverter.parseBase64Binary(jsonResponse.getString("value")));
+                        long offset = jsonResponse.getLong("offset");
 
-                    consumeAsync.complete();
+                        assertEquals(topic, kafkaTopic);
+                        assertEquals(sentBody, value);
+                        assertEquals(0L, offset);
+                        assertNotNull(kafkaPartition);
+                        assertNull(key);
+                    });
+                    consume.complete(true);
                 });
 
-        consumeAsync.await();
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
                 });
 
-        deleteAsync.await();
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
     }
 
     @Test
-    public void receiveFromMultipleTopics(TestContext context) {
+    void receiveFromMultipleTopics(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic1 = "receiveSimpleMessage-1";
         String topic2 = "receiveSimpleMessage-2";
         kafkaCluster.createTopic(topic1, 1, 1);
@@ -738,14 +754,12 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         String sentBody1 = "Simple message-1";
         String sentBody2 = "Simple message-2";
 
-        Async send = context.async();
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send::complete, () -> new ProducerRecord<>(topic1, 0, null, sentBody1));
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic1, 0, null, sentBody1));
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send::complete, () -> new ProducerRecord<>(topic2, 0, null, sentBody2));
-        send.await();
-
-        Async creationAsync = context.async();
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic2, 0, null, sentBody2));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -758,27 +772,27 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("name", name);
         json.put("format", "json");
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonArray topics = new JsonArray();
         topics.add(topic1);
         topics.add(topic2);
@@ -786,70 +800,71 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         JsonObject topicsRoot = new JsonObject();
         topicsRoot.put("topics", topics);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
-        subscriberAsync.await();
-
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonArray> response = ar.result();
-                    context.assertEquals(2, response.body().size());
-
-                    for (int i = 0; i < 2; i++) {
-                        JsonObject jsonResponse = response.body().getJsonObject(i);
-
-                        String kafkaTopic = jsonResponse.getString("topic");
-                        int kafkaPartition = jsonResponse.getInteger("partition");
-                        String key = jsonResponse.getString("key");
-                        String value = jsonResponse.getString("value");
-                        long offset = jsonResponse.getLong("offset");
-
-                        context.assertEquals("receiveSimpleMessage-" + (i + 1), kafkaTopic);
-                        context.assertEquals("Simple message-" + (i + 1), value);
-                        context.assertEquals(0L, offset);
-                        context.assertNotNull(kafkaPartition);
-                        context.assertNull(key);
-                    }
-                    consumeAsync.complete();
-                });
-
-        consumeAsync.await();
-
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonArray> response = ar.result();
+                        assertEquals(2, response.body().size());
+
+                        for (int i = 0; i < 2; i++) {
+                            JsonObject jsonResponse = response.body().getJsonObject(i);
+
+                            String kafkaTopic = jsonResponse.getString("topic");
+                            int kafkaPartition = jsonResponse.getInteger("partition");
+                            String key = jsonResponse.getString("key");
+                            String value = jsonResponse.getString("value");
+                            long offset = jsonResponse.getLong("offset");
+
+                            assertEquals("receiveSimpleMessage-" + (i + 1), kafkaTopic);
+                            assertEquals("Simple message-" + (i + 1), value);
+                            assertEquals(0L, offset);
+                            assertNotNull(kafkaPartition);
+                            assertNull(key);
+                        }
+                    });
+                    consume.complete(true);
+                });
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
-                });
-
-        deleteAsync.await();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
+                });
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
     }
 
     @Test
-    public void receiveFromTopicsWithPattern(TestContext context) {
+    void receiveFromTopicsWithPattern(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic1 = "receiveWithPattern-1";
         String topic2 = "receiveWithPattern-2";
         kafkaCluster.createTopic(topic1, 1, 1);
@@ -858,14 +873,14 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         String sentBody1 = "Simple message-1";
         String sentBody2 = "Simple message-2";
 
-        Async send = context.async(2);
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
+        CompletableFuture<Boolean> produce2 = new CompletableFuture<>();
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send::countDown, () -> new ProducerRecord<>(topic1, 0, null, sentBody1));
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic1, 0, null, sentBody1));
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send::countDown, () -> new ProducerRecord<>(topic2, 0, null, sentBody2));
-        send.await(10000);
-
-        Async creationAsync = context.async();
+            () -> produce2.complete(true), () -> new ProducerRecord<>(topic2, 0, null, sentBody2));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        produce2.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -878,106 +893,105 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("name", name);
         json.put("format", "json");
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonObject topicPattern = new JsonObject();
         topicPattern.put("topic_pattern", "receiveWithPattern-\\d");
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicPattern.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicPattern, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
-        subscriberAsync.await();
-
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonArray> response = ar.result();
+                        assertEquals(2, response.body().size());
 
-                    HttpResponse<JsonArray> response = ar.result();
-                    context.assertEquals(2, response.body().size());
+                        for (int i = 0; i < 2; i++) {
+                            JsonObject jsonResponse = response.body().getJsonObject(i);
 
-                    for (int i = 0; i < 2; i++) {
-                        JsonObject jsonResponse = response.body().getJsonObject(i);
+                            String kafkaTopic = jsonResponse.getString("topic");
+                            int kafkaPartition = jsonResponse.getInteger("partition");
+                            String key = jsonResponse.getString("key");
+                            String value = jsonResponse.getString("value");
+                            long offset = jsonResponse.getLong("offset");
 
-                        String kafkaTopic = jsonResponse.getString("topic");
-                        int kafkaPartition = jsonResponse.getInteger("partition");
-                        String key = jsonResponse.getString("key");
-                        String value = jsonResponse.getString("value");
-                        long offset = jsonResponse.getLong("offset");
-
-                        context.assertEquals("receiveWithPattern-" + (i + 1), kafkaTopic);
-                        context.assertEquals("Simple message-" + (i + 1), value);
-                        context.assertEquals(0L, offset);
-                        context.assertNotNull(kafkaPartition);
-                        context.assertNull(key);
-                    }
-                    consumeAsync.complete();
+                            assertEquals("receiveWithPattern-" + (i + 1), kafkaTopic);
+                            assertEquals("Simple message-" + (i + 1), value);
+                            assertEquals(0L, offset);
+                            assertNull(key);
+                        }
+                    });
+                    consume.complete(true);
                 });
 
-        consumeAsync.await();
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
-                });
-
-        deleteAsync.await();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
+                });
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
     }
 
     @Test
-    public void receiveSimpleMessageFromPartition(TestContext context) {
+    void receiveSimpleMessageFromPartition(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "receiveSimpleMessageFromPartition";
         int partition = 1;
         kafkaCluster.createTopic(topic, 2, 1);
 
         String sentBody = "Simple message from partition";
 
-        Async send = context.async();
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send::complete, () -> new ProducerRecord<>(topic, partition, null, sentBody));
-        send.await();
-
-        Async creationAsync = context.async();
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, partition, null, sentBody));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -990,28 +1004,29 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("name", name);
         json.put("format", "json");
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         // create a consumer
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonArray partitions = new JsonArray();
         partitions.add(new JsonObject().put("topic", topic).put("partition", partition));
 
@@ -1023,75 +1038,73 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(partitionsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
 
-        subscriberAsync.await();
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonArray> response = ar.result();
+                        JsonObject jsonResponse = response.body().getJsonObject(0);
 
-                    HttpResponse<JsonArray> response = ar.result();
-                    JsonObject jsonResponse = response.body().getJsonObject(0);
-
-                    String kafkaTopic = jsonResponse.getString("topic");
-                    int kafkaPartition = jsonResponse.getInteger("partition");
-                    String key = jsonResponse.getString("key");
-                    String value = jsonResponse.getString("value");
-                    long offset = jsonResponse.getLong("offset");
-
-                    context.assertEquals(topic, kafkaTopic);
-                    context.assertEquals(sentBody, value);
-                    context.assertEquals(partition, kafkaPartition);
-                    context.assertEquals(0L, offset);
-                    context.assertNull(key);
+                        String kafkaTopic = jsonResponse.getString("topic");
+                        int kafkaPartition = jsonResponse.getInteger("partition");
+                        String key = jsonResponse.getString("key");
+                        String value = jsonResponse.getString("value");
+                        long offset = jsonResponse.getLong("offset");
 
-                    consumeAsync.complete();
+                        assertEquals(topic, kafkaTopic);
+                        assertEquals(sentBody, value);
+                        assertEquals(partition, kafkaPartition);
+                        assertEquals(0L, offset);
+                        assertNull(key);
+                    });
+                    consume.complete(true);
                 });
 
-        consumeAsync.await();
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> assertTrue(ar.succeeded()));
 
                     HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
+                    assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    delete.complete(true);
                 });
-
-        deleteAsync.await();
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
     }
 
     @Test
-    public void receiveSimpleMessageFromMultiplePartitions(TestContext context) {
+    void receiveSimpleMessageFromMultiplePartitions(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "receiveSimpleMessageFromMultiplePartitions";
         kafkaCluster.createTopic(topic, 2, 1);
 
         String sentBody = "Simple message from partition";
 
-        Async send = context.async(2);
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send::countDown, () -> new ProducerRecord<>(topic, 0, null, sentBody));
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, 0, null, sentBody));
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send::countDown, () -> new ProducerRecord<>(topic, 1, null, sentBody));
-        send.await(10000);
-
-        Async creationAsync = context.async();
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, 1, null, sentBody));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -1104,28 +1117,29 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("name", name);
         json.put("format", "json");
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         // create a consumer
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonArray partitions = new JsonArray();
         partitions.add(new JsonObject().put("topic", topic).put("partition", 0));
         partitions.add(new JsonObject().put("topic", topic).put("partition", 1));
@@ -1138,77 +1152,77 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(partitionsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
 
-        subscriberAsync.await();
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonArray> response = ar.result();
-                    context.assertEquals(2, response.body().size());
-
-                    for (int i = 0; i < 2; i++) {
-                        JsonObject jsonResponse = response.body().getJsonObject(i);
-
-                        String kafkaTopic = jsonResponse.getString("topic");
-                        int kafkaPartition = jsonResponse.getInteger("partition");
-                        String key = jsonResponse.getString("key");
-                        String value = jsonResponse.getString("value");
-                        long offset = jsonResponse.getLong("offset");
-
-                        context.assertEquals("receiveSimpleMessageFromMultiplePartitions", kafkaTopic);
-                        context.assertEquals("Simple message from partition", value);
-                        context.assertEquals(0L, offset);
-                        //context.assertNotNull(kafkaPartition);
-                        context.assertEquals(i, kafkaPartition);
-                        context.assertNull(key);
-                    }
-                    consumeAsync.complete();
-                });
-
-        consumeAsync.await();
-
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonArray> response = ar.result();
+                        assertEquals(2, response.body().size());
+
+                        for (int i = 0; i < 2; i++) {
+                            JsonObject jsonResponse = response.body().getJsonObject(i);
+
+                            String kafkaTopic = jsonResponse.getString("topic");
+                            int kafkaPartition = jsonResponse.getInteger("partition");
+                            String key = jsonResponse.getString("key");
+                            String value = jsonResponse.getString("value");
+                            long offset = jsonResponse.getLong("offset");
+
+                            assertEquals("receiveSimpleMessageFromMultiplePartitions", kafkaTopic);
+                            assertEquals("Simple message from partition", value);
+                            assertEquals(0L, offset);
+                            //context.assertNotNull(kafkaPartition);
+                            assertEquals(i, kafkaPartition);
+                            assertNull(key);
+                        }
+                    });
+                    consume.complete(true);
+                });
+
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
-                });
-
-        deleteAsync.await();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
+                });
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
     }
 
     @Test
-    public void commitOffset(TestContext context) {
+    void commitOffset(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "commitOffset";
         kafkaCluster.createTopic(topic, 1, 1);
 
         String sentBody = "Simple message";
 
-        Async send = context.async();
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send::complete, () -> new ProducerRecord<>(topic, 0, null, sentBody));
-        send.await();
-
-        Async creationAsync = context.async();
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, 0, null, sentBody));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -1222,75 +1236,75 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("format", "json");
         json.put("enable.auto.commit", "false");
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonArray topics = new JsonArray();
         topics.add(topic);
 
         JsonObject topicsRoot = new JsonObject();
         topicsRoot.put("topics", topics);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
 
-        subscriberAsync.await();
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonArray> response = ar.result();
-                    JsonObject jsonResponse = response.body().getJsonObject(0);
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonArray> response = ar.result();
+                        JsonObject jsonResponse = response.body().getJsonObject(0);
 
-                    String kafkaTopic = jsonResponse.getString("topic");
-                    int kafkaPartition = jsonResponse.getInteger("partition");
-                    String key = jsonResponse.getString("key");
-                    String value = jsonResponse.getString("value");
-                    long offset = jsonResponse.getLong("offset");
-
-                    context.assertEquals(topic, kafkaTopic);
-                    context.assertEquals(sentBody, value);
-                    context.assertEquals(0L, offset);
-                    context.assertNotNull(kafkaPartition);
-                    context.assertNull(key);
+                        String kafkaTopic = jsonResponse.getString("topic");
+                        int kafkaPartition = jsonResponse.getInteger("partition");
+                        String key = jsonResponse.getString("key");
+                        String value = jsonResponse.getString("value");
+                        long offset = jsonResponse.getLong("offset");
 
-                    consumeAsync.complete();
+                        assertEquals(topic, kafkaTopic);
+                        assertEquals(sentBody, value);
+                        assertEquals(0L, offset);
+                        assertNotNull(kafkaPartition);
+                        assertNull(key);
+                    });
+                    consume.complete(true);
                 });
 
-        consumeAsync.await();
-
-        Async commitAsync = context.async();
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         JsonArray offsets = new JsonArray();
         json = new JsonObject();
@@ -1302,53 +1316,53 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         JsonObject root = new JsonObject();
         root.put("offsets", offsets);
 
+        CompletableFuture<Boolean> commit = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/offsets")
                 .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
 
-                    int code = response.statusCode();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), code);
-                    commitAsync.complete();
+                        int code = response.statusCode();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), code);
+                    });
+                    commit.complete(true);
                 });
 
-        commitAsync.await();
+        commit.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
-                });
-
-        deleteAsync.await();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
+                });
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
     }
 
     @Test
-    public void commitEmptyOffset(TestContext context) {
+    void commitEmptyOffset(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "commitEmptyOffset";
         kafkaCluster.createTopic(topic, 1, 1);
 
         String sentBody = "Simple message";
 
-        Async send = context.async();
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send::complete, () -> new ProducerRecord<>(topic, 0, null, sentBody));
-        send.await();
-
-        Async creationAsync = context.async();
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, 0, null, sentBody));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -1362,114 +1376,113 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("format", "json");
         json.put("enable.auto.commit", "false");
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonArray topics = new JsonArray();
         topics.add(topic);
 
         JsonObject topicsRoot = new JsonObject();
         topicsRoot.put("topics", topics);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
-        subscriberAsync.await();
-
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonArray> response = ar.result();
-                    JsonObject jsonResponse = response.body().getJsonObject(0);
-
-                    String kafkaTopic = jsonResponse.getString("topic");
-                    int kafkaPartition = jsonResponse.getInteger("partition");
-                    String key = jsonResponse.getString("key");
-                    String value = jsonResponse.getString("value");
-                    long offset = jsonResponse.getLong("offset");
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonArray> response = ar.result();
+                        JsonObject jsonResponse = response.body().getJsonObject(0);
 
-                    context.assertEquals(topic, kafkaTopic);
-                    context.assertEquals(sentBody, value);
-                    context.assertEquals(0L, offset);
-                    context.assertNotNull(kafkaPartition);
-                    context.assertNull(key);
+                        String kafkaTopic = jsonResponse.getString("topic");
+                        int kafkaPartition = jsonResponse.getInteger("partition");
+                        String key = jsonResponse.getString("key");
+                        String value = jsonResponse.getString("value");
+                        long offset = jsonResponse.getLong("offset");
 
-                    consumeAsync.complete();
+                        assertEquals(topic, kafkaTopic);
+                        assertEquals(sentBody, value);
+                        assertEquals(0L, offset);
+                        assertNotNull(kafkaPartition);
+                        assertNull(key);
+                    });
+                    consume.complete(true);
                 });
 
-        consumeAsync.await();
-
-        Async commitAsync = context.async();
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> commit = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/offsets")
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<Buffer> response = ar.result();
 
-                    HttpResponse<Buffer> response = ar.result();
-
-                    int code = response.statusCode();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), code);
-                    commitAsync.complete();
+                        int code = response.statusCode();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), code);
+                    });
+                    commit.complete(true);
                 });
 
-        commitAsync.await();
+        commit.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
-                });
-
-        deleteAsync.await();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
+                });
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
     }
 
     @Test
-    public void emptyRecordTest(TestContext context) {
+    void emptyRecordTest(VertxTestContext context) {
         String topic = "emptyRecordTest";
         kafkaCluster.createTopic(topic, 1, 1);
 
-        Async async = context.async();
-
         JsonObject root = new JsonObject();
 
         WebClient client = WebClient.create(vertx);
@@ -1479,44 +1492,40 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    context.assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), response.statusCode());
-                    context.assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), error.getCode());
-
-                    async.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), response.statusCode());
+                        assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), error.getCode());
+                    });
+                    context.completeNow();
                 });
     }
 
     @Test
-    public void invalidRequestTest(TestContext context) {
-        Async async = context.async();
-
+    void invalidRequestTest(VertxTestContext context) {
         WebClient client = WebClient.create(vertx);
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/not-existing-endpoint")
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(null, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    context.assertEquals(HttpResponseStatus.BAD_REQUEST.code(), response.statusCode());
-                    context.assertEquals(HttpResponseStatus.BAD_REQUEST.code(), error.getCode());
-
-                    async.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.BAD_REQUEST.code(), response.statusCode());
+                        assertEquals(HttpResponseStatus.BAD_REQUEST.code(), error.getCode());
+                    });
+                    context.completeNow();
                 });
     }
 
     @Test
-    public void sendToNonExistingPartitionsTest(TestContext context) {
+    void sendToNonExistingPartitionsTest(VertxTestContext context) {
         String kafkaTopic = "sendToNonExistingPartitionsTest";
         kafkaCluster.createTopic(kafkaTopic, 3, 1);
 
-        Async async = context.async();
-
         String value = "Hi, This is kafka bridge";
         int partition = 1000;
 
@@ -1536,28 +1545,27 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
 
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
+                        JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                        assertEquals(1, offsets.size());
 
-                    JsonArray offsets = bridgeResponse.getJsonArray("offsets");
-                    context.assertEquals(1, offsets.size());
-
-                    HttpBridgeError error = HttpBridgeError.fromJson(offsets.getJsonObject(0));
-                    context.assertEquals(HttpResponseStatus.NOT_FOUND.code(), error.getCode());
-                    context.assertEquals("Invalid partition given with record: 1000 is not in the range [0...3).", error.getMessage());
-                    async.complete();
+                        HttpBridgeError error = HttpBridgeError.fromJson(offsets.getJsonObject(0));
+                        assertEquals(HttpResponseStatus.NOT_FOUND.code(), error.getCode());
+                        assertEquals("Invalid partition given with record: 1000 is not in the range [0...3).", error.getMessage());
+                    });
+                    context.completeNow();
                 });
     }
 
-    @Ignore
+    @Disabled
     @Test
-    public void sendToNonExistingTopicTest(TestContext context) {
+    void sendToNonExistingTopicTest(VertxTestContext context) {
         String kafkaTopic = "sendToNonExistingTopicTest";
 
-        Async async = context.async();
-
         String value = "Hi, This is kafka bridge";
         int partition = 1;
 
@@ -1577,33 +1585,34 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
 
-                    JsonArray offsets = bridgeResponse.getJsonArray("offsets");
-                    context.assertEquals(1, offsets.size());
-                    int code = offsets.getJsonObject(0).getInteger("error_code");
-                    String statusMessage = offsets.getJsonObject(0).getString("error");
+                        JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                        assertEquals(1, offsets.size());
+                        int code = offsets.getJsonObject(0).getInteger("error_code");
+                        String statusMessage = offsets.getJsonObject(0).getString("error");
 
-                    context.assertEquals(HttpResponseStatus.NOT_FOUND.code(), code);
-                    context.assertEquals("Topic " + kafkaTopic + " not found", statusMessage);
-                    async.complete();
+                        assertEquals(HttpResponseStatus.NOT_FOUND.code(), code);
+                        assertEquals("Topic " + kafkaTopic + " not found", statusMessage);
+                    });
+                    context.completeNow();
                 });
     }
 
     @Test
-    public void subscriptionConsumerDoesNotExist(TestContext context) {
+    void subscriptionConsumerDoesNotExist(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "subscriptionConsumerDoesNotExist";
         kafkaCluster.createTopic(topic, 1, 1);
 
         String sentBody = "Simple message";
 
-        Async send = context.async();
-        kafkaCluster.useTo().produceStrings(1, send::complete, () ->
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
+        kafkaCluster.useTo().produceStrings(1, () -> produce.complete(true), () ->
                 new ProducerRecord<>(topic, 0, null, sentBody));
-        send.await();
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -1616,37 +1625,35 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("name", name);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonObject subJson = new JsonObject();
         subJson.put("topic", topic);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(subJson.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(subJson, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    context.assertEquals(HttpResponseStatus.NOT_FOUND.code(), response.statusCode());
-                    context.assertEquals(HttpResponseStatus.NOT_FOUND.code(), error.getCode());
-                    context.assertEquals("The specified consumer instance was not found.", error.getMessage());
-
-                    subscriberAsync.complete();
-                });
-        subscriberAsync.await();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.NOT_FOUND.code(), response.statusCode());
+                        assertEquals(HttpResponseStatus.NOT_FOUND.code(), error.getCode());
+                        assertEquals("The specified consumer instance was not found.", error.getMessage());
+                    });
+                    subscribe.complete(true);
+                });
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
     }
 
 
     @Test
-    public void sendToOnePartitionTest(TestContext context) {
+    void sendToOnePartitionTest(VertxTestContext context) {
         String kafkaTopic = "sendToOnePartitionTest";
         kafkaCluster.createTopic(kafkaTopic, 3, 1);
 
-        Async async = context.async();
-
         String value = "Hi, This is kafka bridge";
         int partition = 1;
 
@@ -1665,28 +1672,27 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-
-                    JsonArray offsets = bridgeResponse.getJsonArray("offsets");
-                    context.assertEquals(1, offsets.size());
-                    JsonObject metadata = offsets.getJsonObject(0);
-                    context.assertNotNull(metadata.getInteger("partition"));
-                    context.assertEquals(partition, metadata.getInteger("partition"));
-                    context.assertEquals(0L, metadata.getLong("offset"));
-                    async.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+
+                        JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                        assertEquals(1, offsets.size());
+                        JsonObject metadata = offsets.getJsonObject(0);
+                        assertNotNull(metadata.getInteger("partition"));
+                        assertEquals(partition, metadata.getInteger("partition"));
+                        assertEquals(0L, metadata.getLong("offset"));
+                    });
+                    context.completeNow();
                 });
     }
 
     @Test
-    public void sendToOneStringPartitionTest(TestContext context) {
+    void sendToOneStringPartitionTest(VertxTestContext context) {
         String kafkaTopic = "sendToOneStringPartitionTest";
         kafkaCluster.createTopic(kafkaTopic, 3, 1);
 
-        Async async = context.async();
-
         String value = "Hi, This is kafka bridge";
         String partition = "karel";
 
@@ -1705,24 +1711,23 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    context.assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), response.statusCode());
-                    context.assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), error.getCode());
-                    context.assertEquals("Specified partition is not a valid number", error.getMessage());
-                    async.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), response.statusCode());
+                        assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), error.getCode());
+                        assertEquals("Specified partition is not a valid number", error.getMessage());
+                    });
+                    context.completeNow();
                 });
     }
 
     @Test
-    public void sendToBothPartitionTest(TestContext context) {
+    void sendToBothPartitionTest(VertxTestContext context) {
         String kafkaTopic = "sendToBothPartitionTest";
         kafkaCluster.createTopic(kafkaTopic, 3, 1);
 
-        Async async = context.async();
-
         String value = "Hi, This is kafka bridge";
         int partition = 1;
 
@@ -1742,32 +1747,29 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    context.assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), response.statusCode());
-                    context.assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), error.getCode());
-                    context.assertEquals("Partition specified in body and in request path", error.getMessage());
-                    async.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), response.statusCode());
+                        assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), error.getCode());
+                        assertEquals("Partition specified in body and in request path", error.getMessage());
+                    });
+                    context.completeNow();
                 });
     }
 
     @Test
-    public void consumerAlreadyExistsTest(TestContext context) {
+    void consumerAlreadyExistsTest(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "consumerAlreadyExistsTest";
         kafkaCluster.createTopic(topic, 1, 1);
 
         String sentBody = "Simple message";
 
-        Async send = context.async();
-        kafkaCluster.useTo().produceStrings(1, send::complete, () ->
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
+        kafkaCluster.useTo().produceStrings(1, () -> produce.complete(true), () ->
                 new ProducerRecord<>(topic, 0, null, sentBody));
-        send.await();
-
-        Async creationAsync = context.async();
-        Async creation2Async = context.async();
-        Async creation3Async = context.async();
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -1779,68 +1781,72 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         JsonObject json = new JsonObject();
         json.put("name", name);
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> create2Again = new CompletableFuture<>();
         // create the same consumer again
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    context.assertEquals(HttpResponseStatus.CONFLICT.code(), response.statusCode());
-                    context.assertEquals(HttpResponseStatus.CONFLICT.code(), error.getCode());
-                    context.assertEquals("A consumer instance with the specified name already exists in the Kafka Bridge.", error.getMessage());
-
-                    creation2Async.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.CONFLICT.code(), response.statusCode());
+                        assertEquals(HttpResponseStatus.CONFLICT.code(), error.getCode());
+                        assertEquals("A consumer instance with the specified name already exists in the Kafka Bridge.", error.getMessage());
+                    });
+                    create2Again.complete(true);
                 });
 
-        creation2Async.await();
+        create2Again.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> create3Again = new CompletableFuture<>();
         // create another consumer
-
         json.put("name", name + "diff");
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name + "diff", consumerInstanceId);
-                    context.assertEquals(baseUri + "diff", consumerBaseUri);
-
-                    creation3Async.complete();
-                });
-
-        creation3Async.await();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name + "diff", consumerInstanceId);
+                        assertEquals(baseUri + "diff", consumerBaseUri);
+                    });
+                    create3Again.complete(true);
+                });
+
+        create3Again.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
     }
 
     @Test
-    public void recordsConsumerDoesNotExist(TestContext context) {
+    void recordsConsumerDoesNotExist(VertxTestContext context) {
         WebClient client = WebClient.create(vertx);
 
         String name = "my-kafka-consumer";
@@ -1849,26 +1855,24 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         String baseUri = "http://" + BRIDGE_HOST + ":" + BRIDGE_PORT + "/consumers/" + groupId + "/instances/" + name;
 
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    context.assertEquals(HttpResponseStatus.NOT_FOUND.code(), response.statusCode());
-                    context.assertEquals(HttpResponseStatus.NOT_FOUND.code(), error.getCode());
-                    context.assertEquals("The specified consumer instance was not found.", error.getMessage());
-
-                    consumeAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.NOT_FOUND.code(), response.statusCode());
+                        assertEquals(HttpResponseStatus.NOT_FOUND.code(), error.getCode());
+                        assertEquals("The specified consumer instance was not found.", error.getMessage());
+                    });
+                    context.completeNow();
                 });
     }
 
     @Test
-    public void offsetsConsumerDoesNotExist(TestContext context) {
+    void offsetsConsumerDoesNotExist(VertxTestContext context) {
         WebClient client = WebClient.create(vertx);
 
         String name = "my-kafka-consumer";
@@ -1877,8 +1881,6 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         String baseUri = "http://" + BRIDGE_HOST + ":" + BRIDGE_PORT + "/consumers/" + groupId + "/instances/" + name;
 
         // commit offsets
-        Async commitAsync = context.async();
-
         JsonArray offsets = new JsonArray();
         JsonObject json = new JsonObject();
         json.put("topic", "offsetsConsumerDoesNotExist");
@@ -1894,31 +1896,29 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    context.assertEquals(HttpResponseStatus.NOT_FOUND.code(), response.statusCode());
-                    context.assertEquals(HttpResponseStatus.NOT_FOUND.code(), error.getCode());
-                    context.assertEquals("The specified consumer instance was not found.", error.getMessage());
-
-                    commitAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.NOT_FOUND.code(), response.statusCode());
+                        assertEquals(HttpResponseStatus.NOT_FOUND.code(), error.getCode());
+                        assertEquals("The specified consumer instance was not found.", error.getMessage());
+                    });
+                    context.completeNow();
                 });
     }
 
     @Test
-    public void doNotRespondTooLongMessage(TestContext context) {
+    void doNotRespondTooLongMessage(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "doNotRespondTooLongMessage";
         kafkaCluster.createTopic(topic, 1, 1);
 
         String sentBody = "Simple message";
 
-        Async send = context.async();
-        kafkaCluster.useTo().produceStrings(1, send::complete, () ->
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
+        kafkaCluster.useTo().produceStrings(1, () -> produce.complete(true), () ->
                 new ProducerRecord<>(topic, 0, null, sentBody));
-        send.await();
-
-        Async creationAsync = context.async();
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -1930,99 +1930,100 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         JsonObject json = new JsonObject();
         json.put("name", name);
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonArray topics = new JsonArray();
         topics.add(topic);
 
         JsonObject topicsRoot = new JsonObject();
         topicsRoot.put("topics", topics);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
 
-        subscriberAsync.await();
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?max_bytes=1")
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_BINARY)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    context.assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), response.statusCode());
-                    context.assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), error.getCode());
-                    context.assertEquals("Response exceeds the maximum number of bytes the consumer can receive",
-                            error.getMessage());
-
-                    consumeAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), response.statusCode());
+                        assertEquals(HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), error.getCode());
+                        assertEquals("Response exceeds the maximum number of bytes the consumer can receive",
+                                error.getMessage());
+                    });
+                    consume.complete(true);
                 });
 
-        consumeAsync.await();
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
                 });
 
-        deleteAsync.await();
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
     }
 
     @Test
-    public void seekToOffsetAndReceive(TestContext context) {
+    @SuppressWarnings("checkstyle:MethodLength")
+    void seekToOffsetAndReceive(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "seekToOffsetAndReceive";
         kafkaCluster.createTopic(topic, 2, 1);
 
-        Async batch = context.async(2);
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         AtomicInteger index0 = new AtomicInteger();
         AtomicInteger index1 = new AtomicInteger();
         kafkaCluster.useTo().produce("", 10, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                batch::countDown, () -> new ProducerRecord<>(topic, 0, "key-" + index0.get(), "value-" + index0.getAndIncrement()));
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, 0, "key-" + index0.get(), "value-" + index0.getAndIncrement()));
         kafkaCluster.useTo().produce("", 10, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                batch::countDown, () -> new ProducerRecord<>(topic, 1, "key-" + index1.get(), "value-" + index1.getAndIncrement()));
-        batch.awaitSuccess(10000);
-
-        Async creationAsync = context.async();
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, 1, "key-" + index1.get(), "value-" + index1.getAndIncrement()));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -2035,55 +2036,56 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("name", name);
         json.put("format", "json");
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
-                });
-        creationAsync.await();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
+                });
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonObject topics = new JsonObject();
         topics.put("topics", new JsonArray().add(topic));
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topics.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topics, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
-        subscriberAsync.await();
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> dummy = new CompletableFuture<>();
         // dummy poll for having re-balancing starting
-        Async dummyPoll = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-                    dummyPoll.complete();
+                    assertTrue(ar.succeeded());
+                    dummy.complete(true);
                 });
-        dummyPoll.await();
+        dummy.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> seek = new CompletableFuture<>();
         // seek
-        Async seekAsync = context.async();
-
         JsonArray offsets = new JsonArray();
         offsets.add(new JsonObject().put("topic", topic).put("partition", 0).put("offset", 9));
         offsets.add(new JsonObject().put("topic", topic).put("partition", 1).put("offset", 5));
@@ -2096,82 +2098,81 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    seekAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    seek.complete(true);
                 });
-        seekAsync.await();
+        seek.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    JsonArray body = ar.result().body();
-
-                    // check it read from partition 0, at offset 9, just one message
-                    List<JsonObject> metadata = body.stream()
-                            .map(JsonObject.class::cast)
-                            .filter(jo -> jo.getInteger("partition") == 0 && jo.getLong("offset") == 9)
-                            .collect(Collectors.toList());
-                    context.assertFalse(metadata.isEmpty());
-                    context.assertEquals(1, metadata.size());
-
-                    context.assertEquals(topic, metadata.get(0).getString("topic"));
-                    context.assertEquals("value-9", metadata.get(0).getString("value"));
-                    context.assertEquals("key-9", metadata.get(0).getString("key"));
-
-                    // check it read from partition 1, starting from offset 5, the last 5 messages
-                    metadata = body.stream()
-                            .map(JsonObject.class::cast)
-                            .filter(jo -> jo.getInteger("partition") == 1)
-                            .collect(Collectors.toList());
-                    context.assertFalse(metadata.isEmpty());
-                    context.assertEquals(5, metadata.size());
-
-                    for (int i = 0; i < metadata.size(); i++) {
-                        context.assertEquals(topic, metadata.get(i).getString("topic"));
-                        context.assertEquals("value-" + (i + 5), metadata.get(i).getString("value"));
-                        context.assertEquals("key-" + (i + 5), metadata.get(i).getString("key"));
-                    }
-
-                    consumeAsync.complete();
-                });
-        consumeAsync.await();
-
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        JsonArray body = ar.result().body();
+
+                        // check it read from partition 0, at offset 9, just one message
+                        List<JsonObject> metadata = body.stream()
+                                .map(JsonObject.class::cast)
+                                .filter(jo -> jo.getInteger("partition") == 0 && jo.getLong("offset") == 9)
+                                .collect(Collectors.toList());
+                        assertFalse(metadata.isEmpty());
+                        assertEquals(1, metadata.size());
+
+                        assertEquals(topic, metadata.get(0).getString("topic"));
+                        assertEquals("value-9", metadata.get(0).getString("value"));
+                        assertEquals("key-9", metadata.get(0).getString("key"));
+
+                        // check it read from partition 1, starting from offset 5, the last 5 messages
+                        metadata = body.stream()
+                                .map(JsonObject.class::cast)
+                                .filter(jo -> jo.getInteger("partition") == 1)
+                                .collect(Collectors.toList());
+                        assertFalse(metadata.isEmpty());
+                        assertEquals(5, metadata.size());
+
+                        for (int i = 0; i < metadata.size(); i++) {
+                            assertEquals(topic, metadata.get(i).getString("topic"));
+                            assertEquals("value-" + (i + 5), metadata.get(i).getString("value"));
+                            assertEquals("key-" + (i + 5), metadata.get(i).getString("key"));
+                        }
+                    });
+                    consume.complete(true);
+                });
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
-                });
-        deleteAsync.await();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
+                });
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
     }
 
     @Test
-    public void seekToBeginningAndReceive(TestContext context) {
+    void seekToBeginningAndReceive(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "seekToBeginningAndReceive";
         kafkaCluster.createTopic(topic, 1, 1);
 
-        Async batch = context.async();
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         AtomicInteger index = new AtomicInteger();
-        kafkaCluster.useTo().produceStrings(10, batch::complete,  () ->
-                new ProducerRecord<>(topic, 0, "key-" + index.get(), "value-" + index.getAndIncrement()));
-        batch.awaitSuccess(10000);
-
-        Async creationAsync = context.async();
+        kafkaCluster.useTo().produce("", 10, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, 0, "key-" + index.get(), "value-" + index.getAndIncrement()));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -2183,120 +2184,123 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         JsonObject json = new JsonObject();
         json.put("name", name);
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonObject topics = new JsonObject();
         topics.put("topics", new JsonArray().add(topic));
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topics.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topics, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
 
-        subscriberAsync.await();
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_BINARY)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    JsonArray body = ar.result().body();
-                    context.assertEquals(10, body.size());
-                    consumeAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        JsonArray body = ar.result().body();
+                        assertEquals(10, body.size());
+                    });
+                    consume.complete(true);
                 });
 
-        consumeAsync.await();
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // seek
-        Async seekAsync = context.async();
-
         JsonArray partitions = new JsonArray();
         partitions.add(new JsonObject().put("topic", topic).put("partition", 0));
 
         JsonObject root = new JsonObject();
         root.put("partitions", partitions);
 
+
+        CompletableFuture<Boolean> seek = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/positions/beginning")
                 .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    seekAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    seek.complete(true);
                 });
 
-        seekAsync.await();
+        seek.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consumeSeek = new CompletableFuture<>();
         // consume records
-        Async consumeSeekAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records")
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_BINARY)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    JsonArray body = ar.result().body();
-                    context.assertEquals(10, body.size());
-                    consumeSeekAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        JsonArray body = ar.result().body();
+                        assertEquals(10, body.size());
+                    });
+                    consumeSeek.complete(true);
                 });
 
-        consumeSeekAsync.await();
+        consumeSeek.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
                 });
 
-        deleteAsync.await();
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
     }
 
     @Test
-    public void seekToEndAndReceive(TestContext context) {
+    void seekToEndAndReceive(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "seekToEndAndReceive";
         kafkaCluster.createTopic(topic, 1, 1);
 
-        Async creationAsync = context.async();
-
         WebClient client = WebClient.create(vertx);
 
         String name = "my-kafka-consumer";
@@ -2307,129 +2311,128 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         JsonObject json = new JsonObject();
         json.put("name", name);
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonObject topics = new JsonObject();
         topics.put("topics", new JsonArray().add(topic));
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topics.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topics, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
-        subscriberAsync.await();
-
+        CompletableFuture<Boolean> dummy = new CompletableFuture<>();
         // dummy poll for having re-balancing starting
-        Async dummyPoll = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_BINARY)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-                    dummyPoll.complete();
+                    context.verify(() -> assertTrue(ar.succeeded()));
+                    dummy.complete(true);
                 });
 
-        dummyPoll.await();
+        dummy.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
-        Async batch = context.async();
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         AtomicInteger index = new AtomicInteger();
-        kafkaCluster.useTo().produceStrings(10, batch::complete,  () ->
-                new ProducerRecord<>(topic, 0, "key-" + index.get(), "value-" + index.getAndIncrement()));
-        batch.awaitSuccess(10000);
+        kafkaCluster.useTo().produceStrings(10, () -> produce.complete(true),
+            () -> new ProducerRecord<>(topic, 0, "key-" + index.get(), "value-" + index.getAndIncrement()));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // seek
-        Async seekAsync = context.async();
-
         JsonArray partitions = new JsonArray();
         partitions.add(new JsonObject().put("topic", topic).put("partition", 0));
 
         JsonObject root = new JsonObject();
         root.put("partitions", partitions);
 
+        CompletableFuture<Boolean> seek = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/positions/end")
                 .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    seekAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    seek.complete(true);
                 });
 
-        seekAsync.await();
+        seek.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consumeSeek = new CompletableFuture<>();
         // consume records
-        Async consumeSeekAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records")
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_BINARY)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> assertTrue(ar.succeeded()));
 
                     JsonArray body = ar.result().body();
-                    context.assertEquals(0, body.size());
-                    consumeSeekAsync.complete();
+                    assertEquals(0, body.size());
+                    consumeSeek.complete(true);
                 });
 
-        consumeSeekAsync.await();
+        consumeSeek.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
                 });
 
-        deleteAsync.await();
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
     }
 
     @Test
-    public void doNotReceiveMessageAfterUnsubscribe(TestContext context) {
+    void doNotReceiveMessageAfterUnsubscribe(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "doNotReceiveMessageAfterUnsubscribe";
         kafkaCluster.createTopic(topic, 1, 1);
 
         String sentBody = "Simple message";
 
-        Async send = context.async();
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send::complete, () -> new ProducerRecord<>(topic, 0, null, sentBody));
-        send.await();
-
-        Async creationAsync = context.async();
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, 0, null, sentBody));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -2442,145 +2445,145 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("name", name);
         json.put("format", "json");
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonArray topics = new JsonArray();
         topics.add(topic);
 
         JsonObject topicsRoot = new JsonObject();
         topicsRoot.put("topics", topics);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
 
-        subscriberAsync.await();
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonArray> response = ar.result();
+                        JsonObject jsonResponse = response.body().getJsonObject(0);
 
-                    HttpResponse<JsonArray> response = ar.result();
-                    JsonObject jsonResponse = response.body().getJsonObject(0);
-
-                    String kafkaTopic = jsonResponse.getString("topic");
-                    int kafkaPartition = jsonResponse.getInteger("partition");
-                    String key = jsonResponse.getString("key");
-                    String value = jsonResponse.getString("value");
-                    long offset = jsonResponse.getLong("offset");
-
-                    context.assertEquals(topic, kafkaTopic);
-                    context.assertEquals(sentBody, value);
-                    context.assertEquals(0L, offset);
-                    context.assertNotNull(kafkaPartition);
-                    context.assertNull(key);
+                        String kafkaTopic = jsonResponse.getString("topic");
+                        int kafkaPartition = jsonResponse.getInteger("partition");
+                        String key = jsonResponse.getString("key");
+                        String value = jsonResponse.getString("value");
+                        long offset = jsonResponse.getLong("offset");
 
-                    consumeAsync.complete();
+                        assertEquals(topic, kafkaTopic);
+                        assertEquals(sentBody, value);
+                        assertEquals(0L, offset);
+                        assertNotNull(kafkaPartition);
+                        assertNull(key);
+                    });
+                    consume.complete(true);
                 });
 
-        consumeAsync.await();
-
-        Async unsubscribeAsync = context.async();
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> unsubscribe = new CompletableFuture<>();
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    unsubscribeAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    unsubscribe.complete(true);
                 });
 
-        unsubscribeAsync.await();
+        unsubscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // Send new record
-        Async send2 = context.async();
+        CompletableFuture<Boolean> produce2 = new CompletableFuture<>();
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send2::complete, () -> new ProducerRecord<>(topic, 0, null, sentBody));
-        send2.await();
+            () -> produce2.complete(true), () -> new ProducerRecord<>(topic, 0, null, sentBody));
+        produce2.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // Try to consume after unsubscription
-        Async consumeAsync2 = context.async();
-
+        CompletableFuture<Boolean> consume2 = new CompletableFuture<>();
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    context.assertEquals(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), response.statusCode());
-                    context.assertEquals(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), error.getCode());
-                    context.assertEquals("Consumer is not subscribed to any topics or assigned any partitions", error.getMessage());
-
-                    consumeAsync2.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), response.statusCode());
+                        assertEquals(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), error.getCode());
+                        assertEquals("Consumer is not subscribed to any topics or assigned any partitions", error.getMessage());
+                    });
+                    consume2.complete(true);
                 });
 
-        consumeAsync2.await();
+        consume2.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
                 });
 
-        deleteAsync.await();
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
     }
 
 
     @Test
-    public void unsubscribeConsumerNotFound(TestContext context) {
+    void unsubscribeConsumerNotFound(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "unsubscribeConsumerNotFound";
         kafkaCluster.createTopic(topic, 1, 1);
 
         String sentBody = "Simple message";
 
-        Async send = context.async();
-        kafkaCluster.useTo().produceStrings(1, send::complete, () ->
-                new ProducerRecord<>(topic, 0, null, sentBody));
-        send.await();
-
-        Async creationAsync = context.async();
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
+        kafkaCluster.useTo().produceStrings(1,
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, 0, null, sentBody));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -2592,96 +2595,95 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         JsonObject json = new JsonObject();
         json.put("name", name);
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonArray topics = new JsonArray();
         topics.add(topic);
 
         JsonObject topicsRoot = new JsonObject();
         topicsRoot.put("topics", topics);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
 
-        subscriberAsync.await();
-
-
-        Async unsubscribeAsync = context.async();
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> unsubscribe = new CompletableFuture<>();
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri + "consumer-invalidation" + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    context.assertEquals(HttpResponseStatus.NOT_FOUND.code(), response.statusCode());
-                    context.assertEquals(HttpResponseStatus.NOT_FOUND.code(), error.getCode());
-                    context.assertEquals("The specified consumer instance was not found.", error.getMessage());
-
-                    unsubscribeAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.NOT_FOUND.code(), response.statusCode());
+                        assertEquals(HttpResponseStatus.NOT_FOUND.code(), error.getCode());
+                        assertEquals("The specified consumer instance was not found.", error.getMessage());
+                    });
+                    unsubscribe.complete(true);
                 });
 
-        unsubscribeAsync.await();
+        unsubscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
                 });
 
-        deleteAsync.await();
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
     }
 
     @Test
-    public void formatAndAcceptMismatch(TestContext context) {
+    void formatAndAcceptMismatch(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "formatAndAcceptMismatch";
         kafkaCluster.createTopic(topic, 1, 1);
 
         String sentBody = "Simple message";
 
-        Async send = context.async();
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         kafkaCluster.useTo().produce("", 1, new KafkaJsonSerializer(), new KafkaJsonSerializer(),
-                send::complete, () -> new ProducerRecord<>(topic, 0, null, sentBody));
-        send.await();
-
-        Async creationAsync = context.async();
+            () -> produce.complete(true), () -> new ProducerRecord<>(topic, 0, null, sentBody));
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         WebClient client = WebClient.create(vertx);
 
@@ -2694,90 +2696,90 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("name", name);
         json.put("format", "json");
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonArray topics = new JsonArray();
         topics.add(topic);
 
         JsonObject topicsRoot = new JsonObject();
         topicsRoot.put("topics", topics);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
 
-        subscriberAsync.await();
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_BINARY)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    context.assertEquals(HttpResponseStatus.NOT_ACCEPTABLE.code(), response.statusCode());
-                    context.assertEquals(HttpResponseStatus.NOT_ACCEPTABLE.code(), error.getCode());
-                    context.assertEquals("Consumer format does not match the embedded format requested by the Accept header.", error.getMessage());
-
-                    consumeAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.NOT_ACCEPTABLE.code(), response.statusCode());
+                        assertEquals(HttpResponseStatus.NOT_ACCEPTABLE.code(), error.getCode());
+                        assertEquals("Consumer format does not match the embedded format requested by the Accept header.", error.getMessage());
+                    });
+                    consume.complete(true);
                 });
 
-        consumeAsync.await();
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
                 });
 
-        deleteAsync.await();
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
     }
 
     @Test
-    public void sendReceiveJsonMessage(TestContext context) {
+    void sendReceiveJsonMessage(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
         String topic = "sendReceiveJsonMessage";
         kafkaCluster.createTopic(topic, 1, 1);
 
-        Async async = context.async();
-
         JsonObject sentKey = new JsonObject()
                 .put("f1", "v1")
                 .put("array", new JsonArray().add(1).add(2));
@@ -2798,28 +2800,28 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         WebClient client = WebClient.create(vertx);
 
+        CompletableFuture<Boolean> produce = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/topics/" + topic)
                 .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
                 .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
 
-                    JsonArray offsets = bridgeResponse.getJsonArray("offsets");
-                    context.assertEquals(1, offsets.size());
-                    JsonObject metadata = offsets.getJsonObject(0);
-                    context.assertEquals(0, metadata.getInteger("partition"));
-                    context.assertEquals(0L, metadata.getLong("offset"));
-
-                    async.complete();
+                        JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                        assertEquals(1, offsets.size());
+                        JsonObject metadata = offsets.getJsonObject(0);
+                        assertEquals(0, metadata.getInteger("partition"));
+                        assertEquals(0L, metadata.getLong("offset"));
+                    });
+                    produce.complete(true);
                 });
 
-        async.await();
+        produce.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
-        Async creationAsync = context.async();
 
         String name = "my-kafka-consumer";
         String groupId = "my-group";
@@ -2830,89 +2832,91 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         consumerConfig.put("name", name);
         consumerConfig.put("format", "json");
 
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(consumerConfig.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(consumerConfig, ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    JsonObject bridgeResponse = response.body();
-                    String consumerInstanceId = bridgeResponse.getString("instance_id");
-                    String consumerBaseUri = bridgeResponse.getString("base_uri");
-                    context.assertEquals(name, consumerInstanceId);
-                    context.assertEquals(baseUri, consumerBaseUri);
-                    creationAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        JsonObject bridgeResponse = response.body();
+                        String consumerInstanceId = bridgeResponse.getString("instance_id");
+                        String consumerBaseUri = bridgeResponse.getString("base_uri");
+                        assertEquals(name, consumerInstanceId);
+                        assertEquals(baseUri, consumerBaseUri);
+                    });
+                    create.complete(true);
                 });
 
-        creationAsync.await();
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
         // subscribe to a topic
-        Async subscriberAsync = context.async();
-
         JsonArray topics = new JsonArray();
         topics.add(topic);
 
         JsonObject topicsRoot = new JsonObject();
         topicsRoot.put("topics", topics);
 
+        CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
                 .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
-                    context.assertTrue(ar.succeeded());
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
-                    subscriberAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), ar.result().statusCode());
+                    });
+                    subscribe.complete(true);
                 });
 
-        subscriberAsync.await();
+        subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
         // consume records
-        Async consumeAsync = context.async();
-
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
                 .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonArray> response = ar.result();
-                    JsonObject jsonResponse = response.body().getJsonObject(0);
-
-                    String kafkaTopic = jsonResponse.getString("topic");
-                    int kafkaPartition = jsonResponse.getInteger("partition");
-                    JsonObject key = jsonResponse.getJsonObject("key");
-                    JsonObject value = jsonResponse.getJsonObject("value");
-                    long offset = jsonResponse.getLong("offset");
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonArray> response = ar.result();
+                        JsonObject jsonResponse = response.body().getJsonObject(0);
 
-                    context.assertEquals(topic, kafkaTopic);
-                    context.assertEquals(sentValue, value);
-                    context.assertEquals(0L, offset);
-                    context.assertNotNull(kafkaPartition);
-                    context.assertEquals(sentKey, key);
+                        String kafkaTopic = jsonResponse.getString("topic");
+                        int kafkaPartition = jsonResponse.getInteger("partition");
+                        JsonObject key = jsonResponse.getJsonObject("key");
+                        JsonObject value = jsonResponse.getJsonObject("value");
+                        long offset = jsonResponse.getLong("offset");
 
-                    consumeAsync.complete();
+                        assertEquals(topic, kafkaTopic);
+                        assertEquals(sentValue, value);
+                        assertEquals(0L, offset);
+                        assertNotNull(kafkaPartition);
+                        assertEquals(sentKey, key);
+                    });
+                    consume.complete(true);
                 });
 
-        consumeAsync.await();
+        consume.get(TEST_TIMEOUT, TimeUnit.SECONDS);
 
+        CompletableFuture<Boolean> delete = new CompletableFuture<>();
         // consumer deletion
-        Async deleteAsync = context.async();
-
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
-                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
-                    context.assertTrue(ar.succeeded());
-
-                    HttpResponse<JsonObject> response = ar.result();
-                    context.assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
-
-                    deleteAsync.complete();
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.NO_CONTENT.code(), response.statusCode());
+                    });
+                    delete.complete(true);
                 });
 
-        deleteAsync.await();
+        delete.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+        context.completeNow();
     }
 }
diff --git a/src/test/java/io/strimzi/kafka/bridge/tracker/OffsetTrackerTest.java b/src/test/java/io/strimzi/kafka/bridge/tracker/OffsetTrackerTest.java
index 80b4e10..3b6cf05 100644
--- a/src/test/java/io/strimzi/kafka/bridge/tracker/OffsetTrackerTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/tracker/OffsetTrackerTest.java
@@ -8,8 +8,8 @@ package io.strimzi.kafka.bridge.tracker;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -20,15 +20,15 @@ import java.util.Map.Entry;
 
 import static org.junit.Assert.assertTrue;
 
-public class OffsetTrackerTest {
+class OffsetTrackerTest {
 
     private static final Logger log = LoggerFactory.getLogger(OffsetTrackerTest.class);
 
     private List<ConsumerRecord<String, byte[]>> records = new ArrayList<>();
     private Map<TopicPartition, OffsetAndMetadata> offsets;
 
-    @Before
-    public void before() {
+    @BeforeEach
+    void before() {
 
         this.records.add(new ConsumerRecord<String, byte[]>("my_topic", 0, 0, null, null));
         this.records.add(new ConsumerRecord<String, byte[]>("my_topic", 0, 1, null, null));
@@ -39,7 +39,7 @@ public class OffsetTrackerTest {
     }
 
     @Test
-    public void fullOffsetTrackerOutOfOrder() {
+    void fullOffsetTrackerOutOfOrder() {
 
         OffsetTracker offsetTracker  = new FullOffsetTracker("my_topic");
 
@@ -93,7 +93,7 @@ public class OffsetTrackerTest {
     }
 
     @Test
-    public void fullOffsetTracker() {
+    void fullOffsetTracker() {
 
         OffsetTracker offsetTracker  = new FullOffsetTracker("my_topic");
 
@@ -147,7 +147,7 @@ public class OffsetTrackerTest {
     }
 
     @Test
-    public void simpleOffsetTrackerOutOfOrder() {
+    void simpleOffsetTrackerOutOfOrder() {
 
         OffsetTracker offsetTracker  = new SimpleOffsetTracker("my_topic");
 
@@ -201,7 +201,7 @@ public class OffsetTrackerTest {
     }
 
     @Test
-    public void simpleOffsetTracker() {
+    void simpleOffsetTracker() {
 
         OffsetTracker offsetTracker  = new SimpleOffsetTracker("my_topic");