PNF WF post instantiation configuration

Add PNF ipv4/v6 address for config Deploy
Add Integration test for CreateVcpeResCustService_simplified.bpmn
Modify the PnfEventReadyDmaapClient to avoid one more running thread
Refactor the so-bpmn-infrastructure-flows test resources
Modify the CreateVcpeResCustService_simplified.bpmn
Add ConfigurePnfResource.bpmn
Add Delegates

Issue-ID: SO-1506
Change-Id: Iffb69d1441ef0b485ee8cd3fb5da5f1a35279a95
Signed-off-by: eeginux <henry.xie@est.tech>

6 files changed, 392 insertions, 1 deletions

diff --git a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/ConfigCheckerDelegate.java b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/ConfigCheckerDelegate.java
new file mode 100644
index 0000000000..9b60a05e98
--- /dev/null
+++ b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/ConfigCheckerDelegate.java
@@ -0,0 +1,97 @@
+/*
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 Nordix Foundation.
+ *  ================================================================================
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.so.bpmn.infrastructure.pnf.delegate;
+
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.MODEL_UUID;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PRC_BLUEPRINT_NAME;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PRC_BLUEPRINT_VERSION;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PRC_CUSTOMIZATION_UUID;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PRC_INSTANCE_NAME;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.SERVICE_MODEL_INFO;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.SKIP_POST_INSTANTIATION_CONFIGURATION;
+
+import java.util.List;
+import org.camunda.bpm.engine.delegate.DelegateExecution;
+import org.camunda.bpm.engine.delegate.JavaDelegate;
+import org.onap.so.bpmn.core.json.JsonUtils;
+import org.onap.so.client.exception.ExceptionBuilder;
+import org.onap.so.db.catalog.beans.PnfResourceCustomization;
+import org.onap.so.db.catalog.client.CatalogDbClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+/**
+ * This implementation of {@link JavaDelegate} is used to check the SO catalogdb for PNF config flag.
+ *
+ * It queries the PNF resource customization table for the skip_post_instantiation_configuration for required PNF model
+ * UUID.
+ */
+@Component
+public class ConfigCheckerDelegate implements JavaDelegate {
+
+    private Logger logger = LoggerFactory.getLogger(ConfigCheckerDelegate.class);
+
+    //ERROR CODE for variable not found in the delegation Context
+    private static int ERROR_CODE = 601;
+
+    @Autowired
+    protected ExceptionBuilder exceptionUtil;
+
+    @Autowired
+    protected CatalogDbClient catalogDbClient;
+
+    @Override
+    public void execute(DelegateExecution delegateExecution) throws Exception {
+
+        logger.debug("Running execute block for activity id:{}, name:{}", delegateExecution.getCurrentActivityId(),
+            delegateExecution.getCurrentActivityName());
+
+        if (delegateExecution.hasVariable(SERVICE_MODEL_INFO)) {
+            String serviceModelInfo = (String) delegateExecution.getVariable(SERVICE_MODEL_INFO);
+            String serviceModelUuid = JsonUtils.getJsonValue(serviceModelInfo, MODEL_UUID);
+            delegateExecution.setVariable(MODEL_UUID, serviceModelUuid);
+            List<PnfResourceCustomization> pnfCustomizations = catalogDbClient
+                .getPnfResourceCustomizationByModelUuid(serviceModelUuid);
+            if (pnfCustomizations != null && pnfCustomizations.size() >= 1) {
+                PnfResourceCustomization pnfResourceCustomization = pnfCustomizations.get(0);
+                boolean skipPostInstantiationConfiguration = pnfResourceCustomization.isSkipPostInstConf();
+                delegateExecution
+                    .setVariable(SKIP_POST_INSTANTIATION_CONFIGURATION, skipPostInstantiationConfiguration);
+                delegateExecution.setVariable(PRC_BLUEPRINT_NAME, pnfResourceCustomization.getBlueprintName());
+                delegateExecution.setVariable(PRC_BLUEPRINT_VERSION, pnfResourceCustomization.getBlueprintVersion());
+                delegateExecution
+                    .setVariable(PRC_CUSTOMIZATION_UUID, pnfResourceCustomization.getModelCustomizationUUID());
+                delegateExecution.setVariable(PRC_INSTANCE_NAME, pnfResourceCustomization.getModelInstanceName());
+            } else {
+                logger
+                    .warn("Unable to find the PNF resource customizations of model service UUID: {}", serviceModelUuid);
+                exceptionUtil.buildAndThrowWorkflowException(delegateExecution, ERROR_CODE,
+                    "Unable to find the PNF resource customizations of model service UUID:  " + serviceModelUuid);
+            }
+        } else {
+            logger.warn("Unable to find the parameter: {} in the execution context", SERVICE_MODEL_INFO);
+            exceptionUtil
+                .buildAndThrowWorkflowException(delegateExecution, ERROR_CODE,
+                    "Unable to find parameter " + SERVICE_MODEL_INFO);
+        }
+    }
+}
diff --git a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/ExecutionVariableNames.java b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/ExecutionVariableNames.java
index 7d449e72a8..c16175b8e2 100644
--- a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/ExecutionVariableNames.java
+++ b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/ExecutionVariableNames.java
@@ -34,4 +34,26 @@ public class ExecutionVariableNames {
     public final static String TIMEOUT_FOR_NOTIFICATION = "timeoutForPnfEntryNotification";
     public final static String PNF_UUID = "pnfUuid";
     public final static String SERVICE_INSTANCE_ID = "serviceInstanceId";
+    public final static String MSO_REQUEST_ID = "msoRequestId";
+    public final static String MODEL_UUID = "modelUuid";
+
+    public final static String SERVICE_MODEL_INFO = "serviceModelInfo";
+
+    /**
+     * Variable used to contain the {@link org.onap.so.client.cds.beans.AbstractCDSPropertiesBean} json value.
+     */
+    public static final String EXECUTION_OBJECT = "executionObject";
+
+    /**
+     * Variables used to contain the {@link org.onap.so.db.catalog.beans.PnfResourceCustomization} fields.
+     */
+    public static final String PRC_BLUEPRINT_NAME = "PRC_blueprintName";
+    public static final String PRC_BLUEPRINT_VERSION = "PRC_blueprintVersion";
+    public static final String PRC_CUSTOMIZATION_UUID = "PRC_customizationUuid";
+    public static final String PRC_INSTANCE_NAME = "PRC_instanceName";
+
+    /**
+     * Variable used to contain skipPostInstantiationConfiguration flag.
+     */
+    public static final String SKIP_POST_INSTANTIATION_CONFIGURATION = "SkipPostInstantiationConfiguration";
 }
diff --git a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/PrepareCdsCallDelegate.java b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/PrepareCdsCallDelegate.java
new file mode 100644
index 0000000000..49887af4be
--- /dev/null
+++ b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/PrepareCdsCallDelegate.java
@@ -0,0 +1,84 @@
+/*
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 Nordix Foundation.
+ *  ================================================================================
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.so.bpmn.infrastructure.pnf.delegate;
+
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.EXECUTION_OBJECT;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.MSO_REQUEST_ID;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PNF_UUID;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PRC_BLUEPRINT_NAME;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PRC_BLUEPRINT_VERSION;
+
+import org.camunda.bpm.engine.delegate.DelegateExecution;
+import org.camunda.bpm.engine.delegate.JavaDelegate;
+import org.onap.so.client.cds.beans.AbstractCDSPropertiesBean;
+import org.onap.so.client.exception.ExceptionBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+
+/**
+ * Abstract class for preparing CDS call.
+ */
+public abstract class PrepareCdsCallDelegate implements JavaDelegate {
+
+    protected Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    protected static final String ORIGINATOR_ID = "SO";
+    protected static final int ERROR_CODE = 7010;
+
+    protected String actionName;
+    protected String mode;
+
+    @Autowired
+    protected ExceptionBuilder exceptionUtil;
+
+    @Override
+    public void execute(DelegateExecution delegateExecution){
+
+        logger.debug("Running execute block for activity:{}", delegateExecution.getCurrentActivityId());
+        AbstractCDSPropertiesBean cdsPropertiesBean = new AbstractCDSPropertiesBean();
+        cdsPropertiesBean.setBlueprintName((String) delegateExecution.getVariable(PRC_BLUEPRINT_NAME));
+        cdsPropertiesBean.setBlueprintVersion((String) delegateExecution.getVariable(PRC_BLUEPRINT_VERSION));
+        cdsPropertiesBean.setOriginatorId(ORIGINATOR_ID);
+        cdsPropertiesBean.setActionName(getActionName());
+        cdsPropertiesBean.setMode(getMode());
+        cdsPropertiesBean.setRequestId((String) delegateExecution.getVariable(MSO_REQUEST_ID));
+        cdsPropertiesBean.setSubRequestId((String) delegateExecution.getVariable(PNF_UUID));
+        cdsPropertiesBean.setRequestObject(getRequestObject(delegateExecution));
+        delegateExecution.setVariable(EXECUTION_OBJECT, cdsPropertiesBean);
+    }
+
+    /**
+     * Return the request object sent to CDS call.
+     *
+     * @param delegateExecution BPMN delegateExecution context
+     * @return string value of the request object
+     */
+    protected abstract String getRequestObject(final DelegateExecution delegateExecution);
+
+    public String getActionName() {
+        return actionName;
+    }
+
+    public String getMode() {
+        return mode;
+    }
+
+}
diff --git a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/PrepareConfigAssignDelegate.java b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/PrepareConfigAssignDelegate.java
new file mode 100644
index 0000000000..5bf167ea7a
--- /dev/null
+++ b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/PrepareConfigAssignDelegate.java
@@ -0,0 +1,79 @@
+/*
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 Nordix Foundation.
+ *  ================================================================================
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.so.bpmn.infrastructure.pnf.delegate;
+
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.MODEL_UUID;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PNF_CORRELATION_ID;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PNF_UUID;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PRC_CUSTOMIZATION_UUID;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PRC_INSTANCE_NAME;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.SERVICE_INSTANCE_ID;
+
+import org.camunda.bpm.engine.delegate.DelegateExecution;
+import org.camunda.bpm.engine.delegate.JavaDelegate;
+import org.onap.so.client.cds.beans.AbstractCDSPropertiesBean;
+import org.onap.so.client.cds.beans.ConfigAssignPropertiesForPnf;
+import org.onap.so.client.cds.beans.ConfigAssignRequestPnf;
+import org.springframework.stereotype.Component;
+
+/**
+ * This implementation of {@link JavaDelegate} is used to prepare for config Assign.
+ *
+ * It queries the PNF resource customization table and construct the {@link AbstractCDSPropertiesBean} as
+ * executionObject.
+ */
+@Component
+public class PrepareConfigAssignDelegate extends PrepareCdsCallDelegate {
+
+    public PrepareConfigAssignDelegate() {
+        this.actionName = "config-assign";
+        this.mode = "sync";
+    }
+
+    @Override
+    protected String getRequestObject(final DelegateExecution delegateExecution){
+
+        ConfigAssignPropertiesForPnf configAssignProperties = new ConfigAssignPropertiesForPnf();
+        configAssignProperties.setServiceInstanceId((String) delegateExecution.getVariable(SERVICE_INSTANCE_ID));
+
+        /**
+         * PNF Name matches the name in AAI, i.e., correlationID as in customized workflow.
+         */
+        configAssignProperties.setPnfName((String) delegateExecution.getVariable(PNF_CORRELATION_ID));
+
+        /**
+         * PNF id match AAI entry, i.e, PNF UUID.
+         */
+        configAssignProperties.setPnfId((String) delegateExecution.getVariable(PNF_UUID));
+        configAssignProperties.setPnfCustomizationUuid((String) delegateExecution.getVariable(PRC_CUSTOMIZATION_UUID));
+        configAssignProperties.setServiceModelUuid((String) delegateExecution.getVariable(MODEL_UUID));
+
+        ConfigAssignRequestPnf configAssignRequest = new ConfigAssignRequestPnf();
+        configAssignRequest.setConfigAssignPropertiesForPnf(configAssignProperties);
+
+        /**
+         * resolution key is the same as PNF name.
+         */
+        configAssignRequest.setResolutionKey((String) delegateExecution.getVariable(PNF_CORRELATION_ID));
+
+        return configAssignRequest.toString();
+    }
+
+}
diff --git a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/PrepareConfigDeployDelegate.java b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/PrepareConfigDeployDelegate.java
new file mode 100644
index 0000000000..9b13be8e52
--- /dev/null
+++ b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/delegate/PrepareConfigDeployDelegate.java
@@ -0,0 +1,108 @@
+/*
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 Nordix Foundation.
+ *  ================================================================================
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.so.bpmn.infrastructure.pnf.delegate;
+
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.MODEL_UUID;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PNF_CORRELATION_ID;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PNF_UUID;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.PRC_CUSTOMIZATION_UUID;
+import static org.onap.so.bpmn.infrastructure.pnf.delegate.ExecutionVariableNames.SERVICE_INSTANCE_ID;
+
+import java.io.IOException;
+import java.util.Optional;
+import org.camunda.bpm.engine.delegate.DelegateExecution;
+import org.camunda.bpm.engine.delegate.JavaDelegate;
+import org.onap.aai.domain.yang.Pnf;
+import org.onap.so.bpmn.infrastructure.pnf.management.PnfManagement;
+import org.onap.so.client.cds.beans.AbstractCDSPropertiesBean;
+import org.onap.so.client.cds.beans.ConfigDeployPropertiesForPnf;
+import org.onap.so.client.cds.beans.ConfigDeployRequestPnf;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+/**
+ * This implementation of {@link JavaDelegate} is used to prepare for config Deploy.
+ *
+ * It queries the PNF resource customization table and construct the {@link AbstractCDSPropertiesBean} as
+ * executionObject.
+ */
+@Component
+public class PrepareConfigDeployDelegate extends PrepareCdsCallDelegate {
+
+    @Autowired
+    private PnfManagement pnfManagement;
+
+    public PrepareConfigDeployDelegate() {
+        this.actionName = "config-deploy";
+        this.mode = "async";
+    }
+
+    @Override
+    protected String getRequestObject(DelegateExecution delegateExecution){
+
+        ConfigDeployPropertiesForPnf configDeployProperties = new ConfigDeployPropertiesForPnf();
+
+        configDeployProperties.setServiceInstanceId((String) delegateExecution.getVariable(SERVICE_INSTANCE_ID));
+
+        /**
+         * PNF Name matches the name in AAI, i.e., correlationID as in customized workflow.
+         */
+        configDeployProperties.setPnfName((String) delegateExecution.getVariable(PNF_CORRELATION_ID));
+
+        /**
+         * PNF id match AAI entry, i.e, PNF UUID.
+         */
+        configDeployProperties.setPnfId((String) delegateExecution.getVariable(PNF_UUID));
+        configDeployProperties.setPnfCustomizationUuid((String) delegateExecution.getVariable(PRC_CUSTOMIZATION_UUID));
+        configDeployProperties.setServiceModelUuid((String) delegateExecution.getVariable(MODEL_UUID));
+        setIpAddress(configDeployProperties, delegateExecution);
+
+        ConfigDeployRequestPnf configDeployRequest = new ConfigDeployRequestPnf();
+        configDeployRequest.setConfigDeployPropertiesForPnf(configDeployProperties);
+
+        /**
+         * Resolution key is the same as PNF name.
+         */
+        configDeployRequest.setResolutionKey((String) delegateExecution.getVariable(PNF_CORRELATION_ID));
+
+        return configDeployRequest.toString();
+    }
+
+    private void setIpAddress(ConfigDeployPropertiesForPnf configDeployProperties, DelegateExecution delegateExecution) {
+
+        /**
+         * Retrieve PNF entry from AAI.
+         */
+        try {
+            String pnfName = (String) delegateExecution.getVariable(PNF_CORRELATION_ID);
+            Optional<Pnf> pnfOptional = pnfManagement.getEntryFor(pnfName);
+            if ( pnfOptional.isPresent()){
+                Pnf pnf = pnfOptional.get();
+                configDeployProperties.setPnfIpV4Address(pnf.getPnfIpv4Address());
+                configDeployProperties.setPnfIpV6Address(pnf.getPnfIpv6Address());
+            } else {
+                exceptionUtil.buildAndThrowWorkflowException(delegateExecution, ERROR_CODE, "AAI entry for PNF: " + pnfName + " does not exist");
+            }
+        } catch (IOException e) {
+            logger.warn(e.getMessage(), e);
+            exceptionUtil.buildAndThrowWorkflowException(delegateExecution, ERROR_CODE, "Unable to fetch from AAI" + e.getMessage());
+        }
+    }
+}
diff --git a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClient.java b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClient.java
index 1b7a69eec8..6f828f6cb0 100644
--- a/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClient.java
+++ b/bpmn/so-bpmn-infrastructure-common/src/main/java/org/onap/so/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClient.java
@@ -136,7 +136,8 @@ public class PnfEventReadyDmaapClient implements DmaapClient {
             Runnable runnable = unregister(pnfCorrelationId);
             if (runnable != null) {
                 logger.debug("dmaap listener gets pnf ready event for pnfCorrelationId: {}", pnfCorrelationId);
-                runnable.run();
+                //runnable.run();
+                runnable = null;
             }
         }
     }