Remove usage of Vertx.currentContext() (#1055)

Fix errors like:

```
\"io.vertx.core.Vertx.currentContext()\" is null\n\tat dev.knative.eventing.kafka.broker.core.security.KubernetesAuthProvider.getCredentials(KubernetesAuthProvider.java:38)
```

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/AuthProvider.java

@@ -18,15 +18,16 @@
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import io.fabric8.kubernetes.client.DefaultKubernetesClient;
 import io.vertx.core.Future;
+import io.vertx.core.Vertx;
 
 /**
  * AuthProvider provides auth credentials.
  */
 @FunctionalInterface
 public interface AuthProvider {
 
-    static AuthProvider kubernetes() {
-        return new KubernetesAuthProvider(new DefaultKubernetesClient());
+    static AuthProvider kubernetes(final Vertx vertx) {
+        return new KubernetesAuthProvider(vertx, new DefaultKubernetesClient());
     }
 
     static AuthProvider noAuth() {

data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesAuthProvider.java

@@ -28,14 +28,16 @@
 
 class KubernetesAuthProvider implements AuthProvider {
 
+    private final Vertx vertx;
     private final KubernetesClient kubernetesClient;
 
-    KubernetesAuthProvider(final KubernetesClient client) {
+    KubernetesAuthProvider(final Vertx vertx, final KubernetesClient client) {
+        this.vertx = vertx;
         this.kubernetesClient = client;
     }
 
     private Future<Credentials> getCredentials(final DataPlaneContract.Reference secretReference) {
-        return Vertx.currentContext().executeBlocking(p -> {
+        return this.vertx.executeBlocking(p -> {
             try {
                 final Secret secret = getSecretFromKubernetes(secretReference);
                 final var credentials = new KubernetesCredentials(secret);
@@ -52,7 +54,7 @@ private Future<Credentials> getCredentials(final DataPlaneContract.Reference sec
     }
 
     private Future<Credentials> getCredentials(final DataPlaneContract.MultiSecretReference secretReferences) {
-        return Vertx.currentContext().executeBlocking(p -> {
+        return this.vertx.executeBlocking(p -> {
             try {
                 final var credentials = new KubernetesCredentials(secretDataOf(secretReferences));
                 final var error = CredentialsValidator.validate(credentials);

data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/KubernetesAuthProviderTest.java

@@ -51,8 +51,8 @@ public class KubernetesAuthProviderTest {
     private static final String userPasswordSecretKey = "password";
 
     @BeforeEach
-    public void setUp() {
-        provider = new KubernetesAuthProvider(client);
+    public void setUp(final Vertx vertx) {
+        provider = new KubernetesAuthProvider(vertx, client);
     }
 
     @Test

data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/Main.java

@@ -113,7 +113,7 @@ public static void start(
                             consumerConfig,
                             webClientOptions,
                             producerConfig,
-                            AuthProvider.kubernetes(),
+                            AuthProvider.kubernetes(vertx),
                             Metrics.getRegistry(),
                             reactiveConsumerFactory,
                             reactiveProducerFactory),

data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverVerticleFactory.java

@@ -69,7 +69,7 @@ public Verticle get() {
                 httpServerOptions,
                 httpsServerOptions,
                 v -> new IngressProducerReconcilableStore(
-                        AuthProvider.kubernetes(),
+                        AuthProvider.kubernetes(v),
                         producerConfigs,
                         properties -> kafkaProducerFactory.create(v, properties)),
                 this.ingressRequestHandler,