1 files changed, 12 insertions, 1 deletions
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java
index e90b0271..5a5eb075 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasksRunner.java
@@ -25,6 +25,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ScheduledFuture;
 import javax.annotation.PreDestroy;
+import org.onap.dcaegen2.services.prh.configuration.CbsConfiguration;
 import org.onap.dcaegen2.services.prh.configuration.PrhProperties;
 import org.springframework.boot.context.event.ApplicationStartedEvent;
 import org.springframework.context.annotation.Configuration;
@@ -46,11 +47,13 @@ public class ScheduledTasksRunner {
     private final TaskScheduler taskScheduler;
     private final ScheduledTasks scheduledTask;
     private final PrhProperties prhProperties;
+    private final CbsConfiguration cbsConfiguration;
     public ScheduledTasksRunner(TaskScheduler taskScheduler, ScheduledTasks scheduledTask,
-        PrhProperties prhProperties) {
+        PrhProperties prhProperties, CbsConfiguration cbsConfiguration) {
         this.taskScheduler = taskScheduler;
         this.scheduledTask = scheduledTask;
         this.prhProperties = prhProperties;
+        this.cbsConfiguration = cbsConfiguration;
     }
     
     @EventListener
@@ -82,4 +85,12 @@ public class ScheduledTasksRunner {
             return false;
         }
     }
+    
+    /**
+     * Function for cleaning resources for kafka subscriber and publisher.
+     */
+    public synchronized void closeKafkaPublisherSubscriber() {
+        cbsConfiguration.getMessageRouterSubscriber().close();
+        cbsConfiguration.getMessageRouterPublisher().close();
+    }
 }