diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/AsyncCloseable.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/AsyncCloseable.java
index 8a40032993..8361748021 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/AsyncCloseable.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/AsyncCloseable.java
@@ -21,7 +21,9 @@
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
+
import java.util.Arrays;
+import java.util.Objects;
import java.util.stream.Collectors;
/**
@@ -53,7 +55,10 @@ static AutoCloseable toAutoCloseable(AsyncCloseable closeable) {
}
/**
- * Compose several {@link AsyncCloseable} into a single {@link AsyncCloseable}. One close failure will cause the whole close to fail.
+ * Compose several {@link AsyncCloseable}s into a single {@link AsyncCloseable}.
+ * One close failure will cause the whole close to fail.
+ *
+ * It filters null futures returned by individual {@link AsyncCloseable} on close.
*
* @param closeables the closeables to compose
* @return the composed closeables
@@ -61,7 +66,9 @@ static AutoCloseable toAutoCloseable(AsyncCloseable closeable) {
static AsyncCloseable compose(AsyncCloseable... closeables) {
return () -> CompositeFuture.all(
Arrays.stream(closeables)
+ .filter(Objects::nonNull)
.map(AsyncCloseable::close)
+ .filter(Objects::nonNull)
.collect(Collectors.toList())
).mapEmpty();
}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java
index ec5d4745bf..7898c15ff5 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java
@@ -15,13 +15,13 @@
*/
package dev.knative.eventing.kafka.broker.dispatcher;
-import io.vertx.core.Future;
+import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
/**
* This class contains hooks for listening events through the {@link dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher} lifecycle.
*/
-public interface RecordDispatcherListener {
+public interface RecordDispatcherListener extends AsyncCloseable {
/**
* The given record has been received.
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
index 7d059f4618..7e9757eae1 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
@@ -81,7 +81,7 @@ public RecordDispatcherImpl(
this.subscriberSender = composeSenderAndSinkHandler(subscriberSender, responseHandler, "subscriber");
this.dlsSender = composeSenderAndSinkHandler(deadLetterSinkSender, responseHandler, "dead letter sink");
this.recordDispatcherListener = recordDispatcherListener;
- this.closeable = AsyncCloseable.compose(responseHandler, deadLetterSinkSender, subscriberSender);
+ this.closeable = AsyncCloseable.compose(responseHandler, deadLetterSinkSender, subscriberSender, recordDispatcherListener);
this.consumerTracer = consumerTracer;
}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java
index 12fac0eb29..572f94f2bf 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/BaseConsumerVerticle.java
@@ -67,11 +67,15 @@ public void start(Promise startPromise) {
public void stop(Promise stopPromise) {
logger.info("Stopping consumer");
- AsyncCloseable.compose(
- this.consumer::close,
- this.recordDispatcher,
- this.closeable
- ).close(stopPromise);
+ final Promise dependenciesClosedPromise = Promise.promise();
+
+ // Close consumer after other objects have been closed.
+ dependenciesClosedPromise.future()
+ .onComplete(r -> this.consumer.close().onComplete(stopPromise));
+
+ AsyncCloseable
+ .compose(this.recordDispatcher, this.closeable)
+ .close(dependenciesClosedPromise);
}
public void setConsumer(KafkaConsumer