Fix ExecutionQueue cannot be cleaned

Fix that LCM Requests in Execution Queue continue after Stop OAM
request sent.
Fix that Appc-Ansible bundle cannot be fully stopped because
of NPE.

Issue-Id: APPC-159
Change-Id: I8f0a3a79a5c572ad84e66f71b4ddb47118704302
Signed-off-by: Hao Kuang <Hao.Kuang@amdocs.com>

2 files changed, 49 insertions, 35 deletions

diff --git a/appc-dispatcher/appc-command-executor/appc-command-executor-core/src/main/java/org/openecomp/appc/executor/impl/CommandExecutorImpl.java b/appc-dispatcher/appc-command-executor/appc-command-executor-core/src/main/java/org/openecomp/appc/executor/impl/CommandExecutorImpl.java
index 5054d34c7..f7ffdadce 100644
--- a/appc-dispatcher/appc-command-executor/appc-command-executor-core/src/main/java/org/openecomp/appc/executor/impl/CommandExecutorImpl.java
+++ b/appc-dispatcher/appc-command-executor/appc-command-executor-core/src/main/java/org/openecomp/appc/executor/impl/CommandExecutorImpl.java
@@ -28,45 +28,50 @@
 package org.openecomp.appc.executor.impl;
 
 
-import java.time.Instant;
-import java.time.temporal.ChronoUnit;
-import java.util.concurrent.TimeUnit;
-
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
 import org.apache.commons.lang.ObjectUtils;
 import org.openecomp.appc.domainmodel.lcm.RuntimeContext;
-import org.openecomp.appc.domainmodel.lcm.ActionLevel;
 import org.openecomp.appc.exceptions.APPCException;
 import org.openecomp.appc.executionqueue.ExecutionQueueService;
-import org.openecomp.appc.executionqueue.impl.ExecutionQueueServiceFactory;
 import org.openecomp.appc.executor.CommandExecutor;
 
-import com.att.eelf.configuration.EELFLogger;
-import com.att.eelf.configuration.EELFManager;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.concurrent.TimeUnit;
 
 
 public class CommandExecutorImpl implements CommandExecutor {
 
-    private CommandTaskFactory executionTaskFactory ;
+    private CommandTaskFactory executionTaskFactory;
     private static final EELFLogger logger = EELFManager.getInstance().getLogger(CommandExecutorImpl.class);
 
     private ExecutionQueueService executionQueueService;
     private ExpiredMessageHandler expiredMessageHandler;
 
-    public CommandExecutorImpl(){
+    public CommandExecutorImpl() {
 
     }
 
+    /**
+     * Injected by blueprint
+     *
+     * @param executionQueueService
+     */
     public void setExecutionQueueService(ExecutionQueueService executionQueueService) {
         this.executionQueueService = executionQueueService;
     }
 
+    /**
+     * Injected by blueprint
+     * @param expiredMessageHandler
+     */
     public void setExpiredMessageHandler(ExpiredMessageHandler expiredMessageHandler) {
         this.expiredMessageHandler = expiredMessageHandler;
     }
 
     public void initialize() {
         logger.info("initialization started of CommandExecutorImpl");
-        executionQueueService = ExecutionQueueServiceFactory.getExecutionQueueService();
         executionQueueService.registerMessageExpirationListener(expiredMessageHandler);
     }
 
@@ -77,13 +82,14 @@ public class CommandExecutorImpl implements CommandExecutor {
     /**
      * Execute given command
      * Create command request and enqueue it for execution.
+     *
      * @param commandExecutorInput Contains CommandHeader,  command , target Id , payload and conf ID (optional)
      * @throws APPCException in case of error.
      */
     @Override
-    public void executeCommand (RuntimeContext commandExecutorInput) throws APPCException{
+    public void executeCommand(RuntimeContext commandExecutorInput) throws APPCException {
         if (logger.isTraceEnabled()) {
-            logger.trace("Entering to executeCommand with CommandExecutorInput = "+ ObjectUtils.toString(commandExecutorInput));
+            logger.trace("Entering to executeCommand with CommandExecutorInput = " + ObjectUtils.toString(commandExecutorInput));
         }
         enqueRequest(commandExecutorInput);
         if (logger.isTraceEnabled()) {
@@ -91,30 +97,31 @@ public class CommandExecutorImpl implements CommandExecutor {
         }
     }
 
-    private RuntimeContext getCommandRequest(RuntimeContext commandExecutorInput){
+    private RuntimeContext getCommandRequest(RuntimeContext commandExecutorInput) {
         if (logger.isTraceEnabled()) {
-            logger.trace("Entering to getCommandRequest with CommandExecutorInput = "+ ObjectUtils.toString(commandExecutorInput));
+            logger.trace("Entering to getCommandRequest with CommandExecutorInput = " + ObjectUtils.toString(commandExecutorInput));
         }
         RuntimeContext commandRequest;
         commandRequest = commandExecutorInput;
         if (logger.isTraceEnabled()) {
-            logger.trace("Exiting from getCommandRequest with (CommandRequest = "+ ObjectUtils.toString(commandRequest)+")");
+            logger.trace("Exiting from getCommandRequest with (CommandRequest = " + ObjectUtils.toString(commandRequest) + ")");
         }
         return commandRequest;
     }
 
     @SuppressWarnings("unchecked")
-    private void enqueRequest(RuntimeContext request) throws APPCException{
+    private void enqueRequest(RuntimeContext request) throws APPCException {
         if (logger.isTraceEnabled()) {
-            logger.trace("Entering to enqueRequest with CommandRequest = "+ ObjectUtils.toString(request));
+            logger.trace("Entering to enqueRequest with CommandRequest = " + ObjectUtils.toString(request));
         }
         try {
-            String action = request.getRequestContext().getAction().name();
             CommandTask commandTask = executionTaskFactory.getExecutionTask(request);
+
             long remainingTTL = getRemainingTTL(request);
-            executionQueueService.putMessage(commandTask,remainingTTL, TimeUnit.MILLISECONDS);
+
+            executionQueueService.putMessage(commandTask, remainingTTL, TimeUnit.MILLISECONDS);
         } catch (Exception e) {
-            logger.error("Exception: "+e.getMessage());
+            logger.error("Exception: " + e.getMessage());
             throw new APPCException(e);
         }
 
@@ -129,9 +136,9 @@ public class CommandExecutorImpl implements CommandExecutor {
         return ChronoUnit.MILLIS.between(Instant.now(), requestTimestamp.plusSeconds(ttl));
     }
 
-    private CommandTask getMessageExecutor(RuntimeContext request){
+    private CommandTask getMessageExecutor(RuntimeContext request) {
         if (logger.isTraceEnabled()) {
-            logger.trace("Entering to getMessageExecutor with command = "+ request);
+            logger.trace("Entering to getMessageExecutor with command = " + request);
         }
         CommandTask executionTask = executionTaskFactory.getExecutionTask(request);
         if (logger.isTraceEnabled()) {
@@ -139,6 +146,4 @@ public class CommandExecutorImpl implements CommandExecutor {
         }
         return executionTask;
     }
-
-
 }
diff --git a/appc-dispatcher/appc-command-executor/appc-command-executor-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/appc-dispatcher/appc-command-executor/appc-command-executor-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml
index f84e97207..5474dcc6f 100644
--- a/appc-dispatcher/appc-command-executor/appc-command-executor-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml
+++ b/appc-dispatcher/appc-command-executor/appc-command-executor-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml
@@ -31,27 +31,36 @@
            xsi:schemaLocation="http://www.osgi.org/xmlns/blueprint/v1.0.0 http://www.osgi.org/xmlns/blueprint/v1.0.0/blueprint.xsd">
     <bean id="commandExecutorBean" class="org.openecomp.appc.executor.impl.CommandExecutorImpl"
           scope="singleton" init-method="initialize">
+        <property name="executionQueueService" ref="ExecutionQueueServiceRef"/>
         <property name="executionTaskFactory" ref="CommandExecutionTaskFactory"/>
         <property name="expiredMessageHandler" ref="expiredMessageHandlerBean"/>
     </bean>
 
-    <bean id="CommandExecutionTaskFactory" class="org.openecomp.appc.executor.impl.CommandTaskFactory" scope="singleton" >
-        <property name="vnfRequestHandler" ref="vnfRequestHandlerService" />
+    <bean id="CommandExecutionTaskFactory" class="org.openecomp.appc.executor.impl.CommandTaskFactory"
+          scope="singleton">
+        <property name="vnfRequestHandler" ref="vnfRequestHandlerService"/>
         <property name="vmRequestHandler" ref="vmRequestHandlerService"/>
-        <property name="workflowManager" ref="WorkFlowManagerRef" />
-        <property name="lifecyclemanager" ref="LifecyclemanagerRef" />
+        <property name="workflowManager" ref="WorkFlowManagerRef"/>
+        <property name="lifecyclemanager" ref="LifecyclemanagerRef"/>
     </bean>
 
     <bean id="expiredMessageHandlerBean" class="org.openecomp.appc.executor.impl.ExpiredMessageHandler"
           scope="singleton">
-        <property name="vnfRequestHandler" ref="vnfRequestHandlerService" />
+        <property name="vnfRequestHandler" ref="vnfRequestHandlerService"/>
         <property name="vmRequestHandler" ref="vmRequestHandlerService"/>
     </bean>
 
-	<reference id="WorkFlowManagerRef" availability="mandatory" activation="eager" interface="org.openecomp.appc.workflow.WorkFlowManager" />
-	<reference id="vnfRequestHandlerService" availability="optional" activation="eager" interface="org.openecomp.appc.requesthandler.RequestHandler" filter="(level=VNF)" />
-    <reference id="vmRequestHandlerService" availability="optional" activation="eager" interface="org.openecomp.appc.requesthandler.RequestHandler" filter="(level=VM)" />
-    <reference id="LifecyclemanagerRef" availability="mandatory" activation="eager" interface="org.openecomp.appc.lifecyclemanager.LifecycleManager" />
+    <reference id="WorkFlowManagerRef" availability="mandatory" activation="eager"
+               interface="org.openecomp.appc.workflow.WorkFlowManager"/>
+    <reference id="vnfRequestHandlerService" availability="optional" activation="eager"
+               interface="org.openecomp.appc.requesthandler.RequestHandler" filter="(level=VNF)"/>
+    <reference id="vmRequestHandlerService" availability="optional" activation="eager"
+               interface="org.openecomp.appc.requesthandler.RequestHandler" filter="(level=VM)"/>
+    <reference id="LifecyclemanagerRef" availability="mandatory" activation="eager"
+               interface="org.openecomp.appc.lifecyclemanager.LifecycleManager"/>
+    <reference id="ExecutionQueueServiceRef" availability="mandatory" activation="eager"
+               interface="org.openecomp.appc.executionqueue.ExecutionQueueService"/>
 
-    <service id="commandExecutorService" interface="org.openecomp.appc.executor.CommandExecutor" ref="commandExecutorBean"/>
+    <service id="commandExecutorService" interface="org.openecomp.appc.executor.CommandExecutor"
+             ref="commandExecutorBean"/>
 </blueprint>