feat: receiver verticle now has a grace period before closing

Signed-off-by: Calum Murray <[email protected]>

Author: Cali0707

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

@@ -97,14 +97,17 @@ public class ReceiverVerticle extends AbstractVerticle implements Handler<HttpSe
     private IngressProducerReconcilableStore ingressProducerStore;
     private FileWatcher secretWatcher;
 
+    private final long terminationGracePeriodMs;
+
     public ReceiverVerticle(
             final ReceiverEnv env,
             final HttpServerOptions httpServerOptions,
             final HttpServerOptions httpsServerOptions,
             final Function<Vertx, IngressProducerReconcilableStore> ingressProducerStoreFactory,
             final IngressRequestHandler ingressRequestHandler,
             final String secretVolumePath,
-            final OIDCDiscoveryConfig oidcDiscoveryConfig) {
+            final OIDCDiscoveryConfig oidcDiscoveryConfig,
+            final long terminationGracePeriodMs) {
 
         Objects.requireNonNull(env);
         Objects.requireNonNull(httpServerOptions);
@@ -122,6 +125,7 @@ public ReceiverVerticle(
         this.tlsKeyFile = new File(secretVolumePath + "/tls.key");
         this.tlsCrtFile = new File(secretVolumePath + "/tls.crt");
         this.oidcDiscoveryConfig = oidcDiscoveryConfig;
+        this.terminationGracePeriodMs = terminationGracePeriodMs;
     }
 
     public HttpServerOptions getHttpsServerOptions() {
@@ -193,6 +197,20 @@ private void setupSecretWatcher() {
 
     @Override
     public void stop(Promise<Void> stopPromise) throws Exception {
+        if (this.terminationGracePeriodMs < 1) {
+            this.handleStop(stopPromise);
+            return;
+        }
+        this.vertx.setTimer(this.terminationGracePeriodMs, (ignored) -> {
+            try {
+                this.handleStop(stopPromise);
+            } catch (Exception e) {
+                stopPromise.fail(e);
+            }
+        });
+    }
+
+    private void handleStop(Promise<Void> stopPromise) throws Exception {
         CompositeFuture.all(
                         (this.httpServer != null ? this.httpServer.close().mapEmpty() : Future.succeededFuture()),
                         (this.httpsServer != null ? this.httpsServer.close().mapEmpty() : Future.succeededFuture()),

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

@@ -68,6 +68,8 @@ public class Main {
 
     private static final Logger logger = LoggerFactory.getLogger(Main.class);
 
+    private static final long receiverGracePeriodMs = 20_000; // 20 seconds, in milliseconds
+
     /**
      * Start receiver.
      *
@@ -165,7 +167,8 @@ public static void start(final String[] args, final ReactiveProducerFactory kafk
                     eventTypeClient,
                     eventTypeLister,
                     vertx,
-                    oidcDiscoveryConfig);
+                    oidcDiscoveryConfig,
+                    receiverGracePeriodMs);
             DeploymentOptions deploymentOptions =
                     new DeploymentOptions().setInstances(Runtime.getRuntime().availableProcessors());
             // Deploy the receiver verticles

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

@@ -52,6 +52,8 @@ class ReceiverVerticleFactory implements Supplier<Verticle> {
 
     private ReactiveProducerFactory<String, CloudEvent> kafkaProducerFactory;
 
+    private final long terminationGracePeriodMs;
+
     ReceiverVerticleFactory(
             final ReceiverEnv env,
             final Properties producerConfigs,
@@ -62,7 +64,8 @@ class ReceiverVerticleFactory implements Supplier<Verticle> {
             final MixedOperation<EventType, KubernetesResourceList<EventType>, Resource<EventType>> eventTypeClient,
             final Lister<EventType> eventTypeLister,
             Vertx vertx,
-            final OIDCDiscoveryConfig oidcDiscoveryConfig)
+            final OIDCDiscoveryConfig oidcDiscoveryConfig,
+            final long terminationGracePeriodMs)
             throws NoSuchAlgorithmException {
         {
             this.env = env;
@@ -75,6 +78,7 @@ class ReceiverVerticleFactory implements Supplier<Verticle> {
                     new EventTypeCreatorImpl(eventTypeClient, eventTypeLister, vertx));
             this.kafkaProducerFactory = kafkaProducerFactory;
             this.oidcDiscoveryConfig = oidcDiscoveryConfig;
+            this.terminationGracePeriodMs = terminationGracePeriodMs;
         }
     }
 
@@ -90,6 +94,7 @@ public Verticle get() {
                         properties -> kafkaProducerFactory.create(v, properties)),
                 this.ingressRequestHandler,
                 secretVolumePath,
-                oidcDiscoveryConfig);
+                oidcDiscoveryConfig,
+                terminationGracePeriodMs);
     }
 }

data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticleTest.java

@@ -151,7 +151,8 @@ public void setUpHTTP(final Vertx vertx, final VertxTestContext testContext) {
                 new IngressRequestHandlerImpl(
                         StrictRequestToRecordMapper.getInstance(), registry, ((event, reference) -> null)),
                 SECRET_VOLUME_PATH,
-                null);
+                null,
+                0);
         vertx.deployVerticle(verticle, testContext.succeeding(ar -> testContext.completeNow()));
 
         // Connect to the logger in ReceiverVerticle

data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticleTracingTest.java

@@ -133,7 +133,8 @@ public void setup() throws ExecutionException, InterruptedException {
                 new IngressRequestHandlerImpl(
                         StrictRequestToRecordMapper.getInstance(), Metrics.getRegistry(), ((event, reference) -> null)),
                 SECRET_VOLUME_PATH,
-                null);
+                null,
+                0);
 
         vertx.deployVerticle(verticle).toCompletionStage().toCompletableFuture().get();
     }

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

@@ -56,7 +56,8 @@ public void shouldCreateMultipleReceiverVerticleInstances(Vertx vertx) throws No
                 mockClient.resources(EventType.class),
                 mock(Lister.class),
                 vertx,
-                mock(OIDCDiscoveryConfig.class));
+                mock(OIDCDiscoveryConfig.class),
+                0);
 
         assertThat(supplier.get()).isNotSameAs(supplier.get());
     }

data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/AbstractDataPlaneTest.java

@@ -396,7 +396,8 @@ private ReceiverVerticle setUpReceiver(final Vertx vertx, final VertxTestContext
                         Metrics.getRegistry(),
                         (((event, reference) -> null))),
                 SECRET_VOLUME_PATH,
-                null);
+                null,
+                0);
 
         final CountDownLatch latch = new CountDownLatch(1);
         vertx.deployVerticle(verticle, context.succeeding(h -> latch.countDown()));