[BUG] Make failed async task report failure on Kafka topic

[cps.git] / cps-ncmp-rest / src / main / java / org / onap / cps / ncmp / rest / executor / CpsNcmpTaskExecutor.java

diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutor.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutor.java
index ba68d5b..2601c7a 100644 (file)
--- a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutor.java
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutor.java
@@ -1,6 +1,6 @@
 /*
  *  ============LICENSE_START=======================================================
- *  Copyright (C) 2022-2023 Nordix Foundation
+ *  Copyright (C) 2022-2024 Nordix Foundation
  *  ================================================================================
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@ package org.onap.cps.ncmp.rest.executor;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
 import java.util.function.Supplier;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -31,16 +32,30 @@ import org.springframework.stereotype.Service;
 @Service
 public class CpsNcmpTaskExecutor {
 
+    /**
+     * Execute a task asynchronously, and invoke completion handler when done.
+     *
+     * @param taskSupplier functional method is get() task needed to be executed asynchronously
+     * @param taskCompletionHandler the action to perform on task completion or error
+     * @param timeOutInMillis the time-out value in milliseconds
+     */
+    public void executeTaskWithErrorHandling(final Supplier<Object> taskSupplier,
+                                             final BiConsumer<Object, Throwable> taskCompletionHandler,
+                                             final long timeOutInMillis) {
+        CompletableFuture.supplyAsync(taskSupplier)
+                .orTimeout(timeOutInMillis, MILLISECONDS)
+                .whenCompleteAsync(taskCompletionHandler);
+    }
+
     /**
      * Execute a task asynchronously.
      *
-     * @param taskSupplier functional method is get() task need to executed asynchronously
+     * @param taskSupplier functional method is get() task needed to be executed asynchronously
      * @param timeOutInMillis the time-out value in milliseconds
      */
     public void executeTask(final Supplier<Object> taskSupplier, final long timeOutInMillis) {
-        CompletableFuture.supplyAsync(taskSupplier::get)
-            .orTimeout(timeOutInMillis, MILLISECONDS)
-            .whenCompleteAsync((taskResult, throwable) -> handleTaskCompletion(throwable));
+        executeTaskWithErrorHandling(taskSupplier, (taskResult, throwable) -> handleTaskCompletion(throwable),
+                timeOutInMillis);
     }
 
     private void handleTaskCompletion(final Throwable throwable) {