From: Jorge Hernandez Date: Tue, 15 Oct 2019 16:19:40 +0000 (+0000) Subject: Merge "Remove actor and recipe checks from ControlLoopCompiler.java" X-Git-Tag: 2.2.0~48 X-Git-Url: https://gerrit.onap.org/r/gitweb?a=commitdiff_plain;h=9c2c12756735b0354de68903bbba7bfb55d56863;hp=bd1e928680cbd46365d7a863c85f5274882a2a51;p=policy%2Fmodels.git Merge "Remove actor and recipe checks from ControlLoopCompiler.java" --- diff --git a/models-examples/src/main/resources/policies/onap.policies.controlloop.guard.MinMax.yaml b/models-examples/src/main/resources/policies/onap.policies.controlloop.guard.MinMax.yaml deleted file mode 100644 index 0fc677d9c..000000000 --- a/models-examples/src/main/resources/policies/onap.policies.controlloop.guard.MinMax.yaml +++ /dev/null @@ -1,43 +0,0 @@ -policy_types: - onap.policies.controlloop.Guard: - derived_from: tosca.policies.Root - version: 1.0.0 - description: Guard Policies for Control Loop Operational Policies - onap.policies.controlloop.guard.MinMax: - derived_from: onap.policies.controlloop.Guard - version: 1.0.0 - description: Supports Min/Max number of VF Modules - properties: - minmax_policy: - type: map - description: null - entry_schema: - type: onap.datatypes.guard.MinMax -data_types: - onap.datatypes.guard.MinMax: - derived_from: tosca.datatypes.Root - properties: - actor: - type: string - description: Specifies the Actor - required: true - recipe: - type: string - description: Specified the Recipe - required: true - time_range: - type: tosca.datatypes.TimeInterval - description: An optional range of time during the day the Min/Max limit is valid for. - required: false - controlLoopName: - type: string - description: An optional specific control loop to apply this guard to. - required: false - min_vf_module_instances: - type: integer - required: true - description: The minimum instances of this VF-Module - max_vf_module_instances: - type: integer - required: false - description: The maximum instances of this VF-Module diff --git a/models-examples/src/main/resources/policies/vCPE.policies.optimization.input.tosca.json b/models-examples/src/main/resources/policies/vCPE.policies.optimization.input.tosca.json new file mode 100644 index 000000000..68e3a4415 --- /dev/null +++ b/models-examples/src/main/resources/policies/vCPE.policies.optimization.input.tosca.json @@ -0,0 +1,234 @@ +{ + "tosca_definitions_version": "tosca_simple_yaml_1_0_0", + "topology_template": { + "policies": [ + { + "OSDF_CASABLANCA.Affinity_vCPE_1": { + "type": "onap.policies.optimization.AffinityPolicy", + "version": "1.0.0", + "metadata": { + "policy-id": "OSDF_CASABLANCA.Affinity_vCPE_1" + }, + "properties": { + "identity": "affinity_vCPE", + "policyScope": [ + "vCPE", + "US", + "INTERNATIONAL", + "ip", + "vGMuxInfra", + "vG" + ], + "affinityProperties": { + "qualifier": "same", + "category": "complex" + }, + "policyType": "zone", + "resources": [ + "vGMuxInfra", + "vG" + ] + } + } + }, + { + "OSDF_CASABLANCA.Capacity_vG_1": { + "type": "onap.policies.optimization.Vim_fit", + "version": "1.0.0", + "metadata": { + "policy-id": "OSDF_CASABLANCA.Capacity_vG_1" + }, + "properties": { + "identity": "capacity_vG", + "policyScope": [ + "VCPE", + "US", + "INTERNATIONAL", + "ip", + "vG" + ], + "resources": [ + "vG" + ], + "capacityProperty": { + "controller": "multicloud", + "request": "{\"vCPU\": 10, \"Memory\": {\"quantity\": {\"get_param\": \"REQUIRED_MEM\"}, \"unit\": \"GB\"}, \"Storage\": {\"quantity\": {\"get_param\": \"REQUIRED_DISK\"}, \"unit\": \"GB\"}}" + }, + "policyType": "vim_fit", + "applicableResources": "any" + } + } + }, + { + "OSDF_CASABLANCA.Distance_vG_1": { + "type": "onap.policies.optimization.DistancePolicy", + "version": "1.0.0", + "metadata": { + "policy-id": "OSDF_CASABLANCA.Distance_vG_1" + }, + "properties": { + "distanceProperties": { + "locationInfo": "customer_loc", + "distance": { + "value": 1500, + "operator": "<", + "unit": "km" + } + }, + "identity": "distance-vG", + "resources": [ + "vG" + ], + "policyScope": [ + "vCPE", + "US", + "INTERNATIONAL", + "ip", + "vG" + ], + "policyType": "distance_to_location", + "applicableResources": "any" + } + } + }, + { + "OSDF_CASABLANCA.hpa_policy_vG_1": { + "type": "onap.policies.optimization.HpaPolicy", + "version": "1.0.0", + "metadata": { + "policy-id": "OSDF_CASABLANCA.hpa_policy_vG_1" + }, + "properties": { + "resources": [ + "vG" + ], + "identity": "hpa-vG", + "policyScope": [ + "vCPE", + "US", + "INTERNATIONAL", + "ip", + "vG" + ], + "policyType": "hpa" + } + } + }, + { + "OSDF_CASABLANCA.queryPolicy_vCPE": { + "type": "onap.policies.optimization.QueryPolicy", + "version": "1.0.0", + "metadata": { + "policy-id": "OSDF_CASABLANCA.queryPolicy_vCPE" + }, + "properties": { + "queryProperties": [ + { + "attribute": "locationId", + "attribute_location": "customerLocation", + "value": "" + }, + { + "attribute": "id", + "attribute_location": "vpnInfo.vpnId", + "value": "" + }, + { + "attribute": "upstreamBW", + "attribute_location": "vpnInfo.upstreamBW", + "value": "" + }, + { + "attribute": "customerLatitude", + "attribute_location": "customerLatitude", + "value": 1.1 + }, + { + "attribute": "customerLongitude", + "attribute_location": "customerLongitude", + "value": 2.2 + } + ], + "serviceName": "vCPE", + "policyScope": [ + "vCPE", + "US", + "INTERNATIONAL", + "ip", + "vGMuxInfra", + "vG" + ], + "policyType": "request_param_query", + "identity": "vCPE_Query_Policy" + } + } + }, + { + "OSDF_CASABLANCA.SubscriberPolicy_v1": { + "type": "onap.policies.optimization.SubscriberPolicy", + "version": "1.0.0", + "metadata": { + "policy-id": "OSDF_CASABLANCA.SubscriberPolicy_v1" + }, + "properties": { + "identity": "subscriber_vCPE", + "policyScope": [ + "vCPE", + "subscriber_x", + "subscriber_y", + "subscriberPolicy" + ], + "properties": { + "subscriberName": [ + "subscriber_x", + "subscriber_y" + ], + "subscriberRole": [ + "PVT Homing" + ], + "provStatus": [ + "CAPPED" + ] + }, + "policyType": "subscriberPolicy", + "serviceName": "vCPE" + } + } + }, + { + "OSDF_CASABLANCA.vnfPolicy_vG": { + "type": "onap.policies.optimization.VnfPolicy", + "version": "1.0.0", + "metadata": { + "policy-id": "OSDF_CASABLANCA.vnfPolicy_vG" + }, + "properties": { + "identity": "vnf_vG", + "policyScope": [ + "vCPE", + "US", + "INTERNATIONAL", + "ip", + "vG" + ], + "policyType": "vnfPolicy", + "resources": [ + "vG" + ], + "applicableResources": "any", + "vnfProperties": [ + { + "inventoryProvider": "aai", + "serviceType": "", + "inventoryType": "cloud", + "customerId": "", + "orchestrationStatus": "", + "equipmentRole": "" + } + ] + } + } + } + ] + } +} \ No newline at end of file diff --git a/models-interactions/model-actors/actor.cds/src/main/java/org/onap/policy/controlloop/actor/cds/beans/CdsActionRequest.java b/models-interactions/model-actors/actor.cds/src/main/java/org/onap/policy/controlloop/actor/cds/beans/CdsActionRequest.java index 8aac67bb0..45fbbca65 100644 --- a/models-interactions/model-actors/actor.cds/src/main/java/org/onap/policy/controlloop/actor/cds/beans/CdsActionRequest.java +++ b/models-interactions/model-actors/actor.cds/src/main/java/org/onap/policy/controlloop/actor/cds/beans/CdsActionRequest.java @@ -19,6 +19,8 @@ package org.onap.policy.controlloop.actor.cds.beans; import com.google.gson.annotations.SerializedName; + +import java.io.Serializable; import java.util.Map; import lombok.Getter; import lombok.Setter; @@ -30,7 +32,8 @@ import org.slf4j.LoggerFactory; @Getter @Setter -public class CdsActionRequest { +public class CdsActionRequest implements Serializable { + private static final long serialVersionUID = -4172157702597791493L; private static final Logger LOGGER = LoggerFactory.getLogger(CdsActionRequest.class); private static final Coder CODER = new StandardCoder(); diff --git a/models-interactions/model-actors/actor.cds/src/main/java/org/onap/policy/controlloop/actor/cds/constants/CdsActorConstants.java b/models-interactions/model-actors/actor.cds/src/main/java/org/onap/policy/controlloop/actor/cds/constants/CdsActorConstants.java index b79aca061..9adb69232 100644 --- a/models-interactions/model-actors/actor.cds/src/main/java/org/onap/policy/controlloop/actor/cds/constants/CdsActorConstants.java +++ b/models-interactions/model-actors/actor.cds/src/main/java/org/onap/policy/controlloop/actor/cds/constants/CdsActorConstants.java @@ -19,7 +19,6 @@ package org.onap.policy.controlloop.actor.cds.constants; public class CdsActorConstants { - public static final String CDS_ACTOR = "CDS"; // CDS Status @@ -36,4 +35,8 @@ public class CdsActorConstants { public static final String ORIGINATOR_ID = "POLICY"; // Temporarily set to synchronous mode to support current rules, since callbacks aren't supported yet public static final String CDS_MODE = "sync"; + + private CdsActorConstants() { + // Hide implicit public constructor + } } diff --git a/models-interactions/model-actors/actor.cds/src/test/java/org/onap/policy/controlloop/actor/cds/CdsActorServiceProviderTest.java b/models-interactions/model-actors/actor.cds/src/test/java/org/onap/policy/controlloop/actor/cds/CdsActorServiceProviderTest.java index 28a1676ed..eb82ae231 100644 --- a/models-interactions/model-actors/actor.cds/src/test/java/org/onap/policy/controlloop/actor/cds/CdsActorServiceProviderTest.java +++ b/models-interactions/model-actors/actor.cds/src/test/java/org/onap/policy/controlloop/actor/cds/CdsActorServiceProviderTest.java @@ -87,6 +87,8 @@ public class CdsActorServiceProviderTest { // Setup policy policy = new Policy(); Map payloadMap = new HashMap() { + private static final long serialVersionUID = 1L; + { put(CdsActorConstants.KEY_CBA_NAME, CDS_BLUEPRINT_NAME); put(CdsActorConstants.KEY_CBA_VERSION, CDS_BLUEPRINT_VERSION); @@ -124,7 +126,7 @@ public class CdsActorServiceProviderTest { @Test public void testActor() { - assertEquals(cdsActor.actor(), CdsActorConstants.CDS_ACTOR); + assertEquals(CdsActorConstants.CDS_ACTOR, cdsActor.actor()); } @Test @@ -147,7 +149,7 @@ public class CdsActorServiceProviderTest { assertTrue(cdsRequest.hasCommonHeader()); CommonHeader commonHeader = cdsRequest.getCommonHeader(); assertEquals(commonHeader.getRequestId(), REQUEST_ID.toString()); - assertEquals(commonHeader.getSubRequestId(), SUBREQUEST_ID); + assertEquals(SUBREQUEST_ID, commonHeader.getSubRequestId()); assertTrue(cdsRequest.hasPayload()); Struct cdsPayload = cdsRequest.getPayload(); @@ -155,24 +157,24 @@ public class CdsActorServiceProviderTest { assertTrue(cdsRequest.hasActionIdentifiers()); ActionIdentifiers actionIdentifiers = cdsRequest.getActionIdentifiers(); - assertEquals(actionIdentifiers.getActionName(), CDS_RECIPE); - assertEquals(actionIdentifiers.getBlueprintName(), CDS_BLUEPRINT_NAME); - assertEquals(actionIdentifiers.getBlueprintVersion(), CDS_BLUEPRINT_VERSION); + assertEquals(CDS_RECIPE, actionIdentifiers.getActionName()); + assertEquals(CDS_BLUEPRINT_NAME, actionIdentifiers.getBlueprintName()); + assertEquals(CDS_BLUEPRINT_VERSION, actionIdentifiers.getBlueprintVersion()); } @Test public void testRecipePayloads() { - assertEquals(cdsActor.recipePayloads("").size(), 0); + assertEquals(0, cdsActor.recipePayloads("").size()); } @Test public void testRecipes() { - assertEquals(cdsActor.recipes().size(), 0); + assertEquals(0, cdsActor.recipes().size()); } @Test public void testRecipeTargets() { - assertEquals(cdsActor.recipeTargets("").size(), 0); + assertEquals(0, cdsActor.recipeTargets("").size()); } @Test diff --git a/models-interactions/model-actors/actor.so/src/main/java/org/onap/policy/controlloop/actor/so/SoActorServiceProvider.java b/models-interactions/model-actors/actor.so/src/main/java/org/onap/policy/controlloop/actor/so/SoActorServiceProvider.java index 8aad5feb3..842d0f6a5 100644 --- a/models-interactions/model-actors/actor.so/src/main/java/org/onap/policy/controlloop/actor/so/SoActorServiceProvider.java +++ b/models-interactions/model-actors/actor.so/src/main/java/org/onap/policy/controlloop/actor/so/SoActorServiceProvider.java @@ -207,6 +207,7 @@ public class SoActorServiceProvider implements Actor { soModelInfo.setModelName(policy.getTarget().getModelName()); soModelInfo.setModelVersion(policy.getTarget().getModelVersion()); soModelInfo.setModelVersionId(policy.getTarget().getModelVersionId()); + soModelInfo.setModelType("vfModule"); return soModelInfo; } diff --git a/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiCqResponse.java b/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiCqResponse.java index ca0c1867f..7024231e3 100644 --- a/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiCqResponse.java +++ b/models-interactions/model-impl/aai/src/main/java/org/onap/policy/aai/AaiCqResponse.java @@ -62,12 +62,17 @@ public class AaiCqResponse { properties.put(JAXBContextProperties.JSON_INCLUDE_ROOT, false); // Define JAXB context try { - jaxbContext = - JAXBContextFactory - .createContext( - new Class[] {Vserver.class, GenericVnf.class, VfModule.class, - CloudRegion.class, ServiceInstance.class, Tenant.class, ModelVer.class}, - properties); + // @formatter:off + jaxbContext = JAXBContextFactory.createContext(new Class[] { + Vserver.class, + GenericVnf.class, + VfModule.class, + CloudRegion.class, + ServiceInstance.class, + Tenant.class, + ModelVer.class + }, properties); + // @formatter:on unmarshaller = jaxbContext.createUnmarshaller(); } catch (JAXBException e) { LOGGER.error("Could not initialize JAXBContext", e); @@ -81,8 +86,7 @@ public class AaiCqResponse { /** * Constructor creates a custom query response from a valid json string. * - * @param jsonString - * A&AI Custom Query response JSON string + * @param jsonString A&AI Custom Query response JSON string */ public AaiCqResponse(String jsonString) { @@ -98,8 +102,8 @@ public class AaiCqResponse { // Create the StreamSource by creating StringReader using the // JSON input - StreamSource json = new StreamSource(new StringReader( - resultsArray.getJSONObject(i).getJSONObject("vserver").toString())); + StreamSource json = new StreamSource( + new StringReader(resultsArray.getJSONObject(i).getJSONObject("vserver").toString())); // Getting the vserver pojo again from the json Vserver vserver = this.getAaiObject(json, Vserver.class); @@ -110,8 +114,8 @@ public class AaiCqResponse { if (resultsArray.getJSONObject(i).has(GENERIC_VNF)) { // Create the StreamSource by creating StringReader using the // JSON input - StreamSource json = new StreamSource(new StringReader( - resultsArray.getJSONObject(i).getJSONObject(GENERIC_VNF).toString())); + StreamSource json = new StreamSource( + new StringReader(resultsArray.getJSONObject(i).getJSONObject(GENERIC_VNF).toString())); // Getting the generic vnf pojo again from the json GenericVnf genericVnf = this.getAaiObject(json, GenericVnf.class); @@ -124,8 +128,8 @@ public class AaiCqResponse { // Create the StreamSource by creating StringReader using the // JSON input - StreamSource json = new StreamSource(new StringReader( - resultsArray.getJSONObject(i).getJSONObject("service-instance").toString())); + StreamSource json = new StreamSource( + new StringReader(resultsArray.getJSONObject(i).getJSONObject("service-instance").toString())); // Getting the employee pojo again from the json ServiceInstance serviceInstance = this.getAaiObject(json, ServiceInstance.class); @@ -137,8 +141,8 @@ public class AaiCqResponse { if (resultsArray.getJSONObject(i).has(VF_MODULE)) { // Create the StreamSource by creating StringReader using the // JSON input - StreamSource json = new StreamSource(new StringReader( - resultsArray.getJSONObject(i).getJSONObject(VF_MODULE).toString())); + StreamSource json = new StreamSource( + new StringReader(resultsArray.getJSONObject(i).getJSONObject(VF_MODULE).toString())); // Getting the vf module pojo again from the json VfModule vfModule = this.getAaiObject(json, VfModule.class); @@ -150,8 +154,8 @@ public class AaiCqResponse { if (resultsArray.getJSONObject(i).has("cloud-region")) { // Create the StreamSource by creating StringReader using the // JSON input - StreamSource json = new StreamSource(new StringReader( - resultsArray.getJSONObject(i).getJSONObject("cloud-region").toString())); + StreamSource json = new StreamSource( + new StringReader(resultsArray.getJSONObject(i).getJSONObject("cloud-region").toString())); // Getting the cloud region pojo again from the json CloudRegion cloudRegion = this.getAaiObject(json, CloudRegion.class); @@ -163,8 +167,8 @@ public class AaiCqResponse { if (resultsArray.getJSONObject(i).has("tenant")) { // Create the StreamSource by creating StringReader using the // JSON input - StreamSource json = new StreamSource(new StringReader( - resultsArray.getJSONObject(i).getJSONObject("tenant").toString())); + StreamSource json = new StreamSource( + new StringReader(resultsArray.getJSONObject(i).getJSONObject("tenant").toString())); // Getting the tenant pojo again from the json Tenant tenant = this.getAaiObject(json, Tenant.class); @@ -176,8 +180,8 @@ public class AaiCqResponse { if (resultsArray.getJSONObject(i).has("model-ver")) { // Create the StreamSource by creating StringReader using the // JSON input - StreamSource json = new StreamSource(new StringReader( - resultsArray.getJSONObject(i).getJSONObject("model-ver").toString())); + StreamSource json = new StreamSource( + new StringReader(resultsArray.getJSONObject(i).getJSONObject("model-ver").toString())); // Getting the ModelVer pojo again from the json ModelVer modelVer = this.getAaiObject(json, ModelVer.class); @@ -209,8 +213,7 @@ public class AaiCqResponse { /** * Get list of A&AI objects in the custom query. * - * @param classOfResponse - * Class of the type of A&AI objects to be returned + * @param classOfResponse Class of the type of A&AI objects to be returned * @return List A&AI objects matching the class */ @SuppressWarnings("unchecked") @@ -292,8 +295,7 @@ public class AaiCqResponse { /** * Returns a generic Vnf matching vnf name. * - * @param vnfName - * Name of the vnf to match + * @param vnfName Name of the vnf to match * @return generic Vnf */ public GenericVnf getGenericVnfByVnfName(String vnfName) { @@ -318,8 +320,7 @@ public class AaiCqResponse { /** * Returns a generic Vnf matching model invariant ID. * - * @param modelInvariantId - * Name of the vnf to match + * @param modelInvariantId Name of the vnf to match * @return generic Vnf */ public GenericVnf getGenericVnfByModelInvariantId(String modelInvariantId) { @@ -344,8 +345,7 @@ public class AaiCqResponse { /** * Returns a generic Vnf of a given VF Module ID. * - * @param vfModuleModelInvariantId - * of the vf module for which vnf is to be returned + * @param vfModuleModelInvariantId of the vf module for which vnf is to be returned * @return generic Vnf */ public GenericVnf getGenericVnfByVfModuleModelInvariantId(String vfModuleModelInvariantId) { @@ -355,7 +355,7 @@ public class AaiCqResponse { // Iterate through all the vfModules of that generic Vnf for (VfModule vfMod : genVnf.getVfModules().getVfModule()) { if (vfMod.getModelInvariantId() != null - && vfMod.getModelInvariantId().equals(vfModuleModelInvariantId)) { + && vfMod.getModelInvariantId().equals(vfModuleModelInvariantId)) { return genVnf; } } @@ -502,8 +502,7 @@ public class AaiCqResponse { VfModule vfModule = null; for (VfModule vfMod : this.getAllVfModules()) { - if (vfMod.getModelInvariantId() != null - && vfModelInvariantId.equals(vfMod.getModelInvariantId())) { + if (vfMod.getModelInvariantId() != null && vfModelInvariantId.equals(vfMod.getModelInvariantId())) { vfModule = vfMod; } @@ -560,28 +559,25 @@ public class AaiCqResponse { } /** - * Get the count of vfModules matching customizationId, InvariantId and - * VersionId. + * Get the count of vfModules matching customizationId, InvariantId and VersionId. * - * @param custId - * ModelCustomizationId - * @param invId - * ModelInvariantId - * @param verId - * ModelVersionId + * @param custId ModelCustomizationId + * @param invId ModelInvariantId + * @param verId ModelVersionId * @return Returns the count of vf modules */ public int getVfModuleCount(String custId, String invId, String verId) { List vfModuleList = this.getAllVfModules(); int count = 0; for (VfModule vfModule : vfModuleList) { - if (vfModule.getModelCustomizationId() != null && vfModule.getModelInvariantId() != null - && vfModule.getModelVersionId() != null) { - if (vfModule.getModelCustomizationId().equals(custId) - && vfModule.getModelInvariantId().equals(invId) + if (vfModule.getModelCustomizationId() == null || vfModule.getModelInvariantId() == null + || vfModule.getModelVersionId() == null) { + continue; + } + + if (vfModule.getModelCustomizationId().equals(custId) && vfModule.getModelInvariantId().equals(invId) && vfModule.getModelVersionId().equals(verId)) { - count = count + 1; - } + count = count + 1; } } return count; diff --git a/models-interactions/model-impl/cds/src/main/java/org/onap/policy/cds/CdsResponse.java b/models-interactions/model-impl/cds/src/main/java/org/onap/policy/cds/CdsResponse.java index fca1aa2e9..499d1a7c9 100644 --- a/models-interactions/model-impl/cds/src/main/java/org/onap/policy/cds/CdsResponse.java +++ b/models-interactions/model-impl/cds/src/main/java/org/onap/policy/cds/CdsResponse.java @@ -31,6 +31,7 @@ import lombok.ToString; @Setter @ToString public class CdsResponse implements Serializable { + private static final long serialVersionUID = 2590429952699969650L; private String requestId; private String status; diff --git a/models-interactions/model-impl/cds/src/main/java/org/onap/policy/cds/api/CdsProcessorListener.java b/models-interactions/model-impl/cds/src/main/java/org/onap/policy/cds/api/CdsProcessorListener.java index c07c559c4..fb4c6ed79 100644 --- a/models-interactions/model-impl/cds/src/main/java/org/onap/policy/cds/api/CdsProcessorListener.java +++ b/models-interactions/model-impl/cds/src/main/java/org/onap/policy/cds/api/CdsProcessorListener.java @@ -26,7 +26,9 @@ import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOu * the received messages appropriately, it needs to implement {@link CdsProcessorListener}. *

* - *

Here is a sample implementation of a listener: + *

+ * Here is a sample implementation of a listener: + * * 

  * new CdsProcessorListener {
  *
@@ -49,11 +51,10 @@ public interface CdsProcessorListener {
     /**
      * Implements the workflow upon receiving the message from the server side.
      *
-     * 
Note that the CDS client-server communication is configured to use a streaming approach, which means when
-     * client
-     * sends an event, the server can reply with multiple sub-responses until full completion of the processing. Hence,
-     * it is up to the implementation of this method to process the received message using {@link
-     * ExecutionServiceOutput#getStatus()#getEventType()}

+     * 
Note that the CDS client-server communication is configured to use a streaming approach, which means when a
+     * client sends an event, the server can reply with multiple sub-responses until full completion of the processing.
+     * Hence, it is up to the implementation of this method to process the received message using the
+     * getStatus().getEventType() method of {@link ExecutionServiceOutput}
      *
      * @param message ExecutionServiceOutput received by the CDS grpc server
      */
diff --git a/models-interactions/model-impl/rest/src/main/java/org/onap/policy/rest/RestManager.java b/models-interactions/model-impl/rest/src/main/java/org/onap/policy/rest/RestManager.java
index 2ee47a9fc..643c629b3 100644
--- a/models-interactions/model-impl/rest/src/main/java/org/onap/policy/rest/RestManager.java
+++ b/models-interactions/model-impl/rest/src/main/java/org/onap/policy/rest/RestManager.java
@@ -41,9 +41,11 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class RestManager {
-
     private static final Logger logger = LoggerFactory.getLogger(RestManager.class);
 
+    // Constants for string literals
+    private static final String CONTENT_TYPE = "Content-Type";
+
     public class Pair {
         public final A first;
         public final B second;
@@ -69,7 +71,7 @@ public class RestManager {
                                       Map headers, String contentType, String body) {
         HttpPut put = new HttpPut(url);
         addHeaders(put, username, password, headers);
-        put.addHeader("Content-Type", contentType);
+        put.addHeader(CONTENT_TYPE, contentType);
         try {
             StringEntity input = new StringEntity(body);
             input.setContentType(contentType);
@@ -96,7 +98,7 @@ public class RestManager {
                                       Map headers, String contentType, String body) {
         HttpPost post = new HttpPost(url);
         addHeaders(post, username, password, headers);
-        post.addHeader("Content-Type", contentType);
+        post.addHeader(CONTENT_TYPE, contentType);
         try {
             StringEntity input = new StringEntity(body);
             input.setContentType(contentType);
@@ -141,7 +143,7 @@ public class RestManager {
         HttpDeleteWithBody delete = new HttpDeleteWithBody(url);
         addHeaders(delete, username, password, headers);
         if (body != null && !body.isEmpty()) {
-            delete.addHeader("Content-Type", contentType);
+            delete.addHeader(CONTENT_TYPE, contentType);
             try {
                 StringEntity input = new StringEntity(body);
                 input.setContentType(contentType);
diff --git a/models-interactions/model-impl/so/src/main/java/org/onap/policy/so/SoManager.java b/models-interactions/model-impl/so/src/main/java/org/onap/policy/so/SoManager.java
index 6e2494b72..d8933c86c 100644
--- a/models-interactions/model-impl/so/src/main/java/org/onap/policy/so/SoManager.java
+++ b/models-interactions/model-impl/so/src/main/java/org/onap/policy/so/SoManager.java
@@ -142,7 +142,7 @@ public final class SoManager {
             final String vfModuleInstanceId,
             final SoRequest request) {
         return executors.submit(new AsyncSoRestCallThread(requestId, callback, serviceInstanceId, vnfInstanceId,
-                vfModuleInstanceId, request, this.url, this.user, this.password));
+                vfModuleInstanceId, request, this));
     }
 
     /**
@@ -173,18 +173,16 @@ public final class SoManager {
                 final SoCallback callback, final String serviceInstanceId,
                 final String vnfInstanceId, final String vfModuleInstanceId,
                 final SoRequest request,
-                final String url,
-                final String user,
-                final String password) {
+                final SoManager callingSoManager) {
             this.requestId = requestId;
             this.callback = callback;
             this.serviceInstanceId = serviceInstanceId;
             this.vnfInstanceId = vnfInstanceId;
             this.vfModuleInstanceId = vfModuleInstanceId;
             this.request = request;
-            this.baseUrl = url;
-            this.user = user;
-            this.password = password;
+            this.baseUrl = callingSoManager.url;
+            this.user = callingSoManager.user;
+            this.password = callingSoManager.password;
         }
 
         /**
diff --git a/models-interactions/model-impl/vfc/src/main/java/org/onap/policy/vfc/VfcManager.java b/models-interactions/model-impl/vfc/src/main/java/org/onap/policy/vfc/VfcManager.java
index 406e35d33..14382f0e9 100644
--- a/models-interactions/model-impl/vfc/src/main/java/org/onap/policy/vfc/VfcManager.java
+++ b/models-interactions/model-impl/vfc/src/main/java/org/onap/policy/vfc/VfcManager.java
@@ -125,45 +125,7 @@ public final class VfcManager implements Runnable {
         }
 
         try {
-            VfcResponse response = Serialization.gsonPretty.fromJson(httpDetails.second, VfcResponse.class);
-            NetLoggerUtil.log(EventType.IN, CommInfrastructure.REST, vfcUrl, httpDetails.second);
-            String body = Serialization.gsonPretty.toJson(response);
-            logger.debug("Response to VFC Heal post:");
-            logger.debug(body);
-
-            String jobId = response.getJobId();
-            int attemptsLeft = 20;
-
-            String urlGet = vfcUrlBase + "/jobs/" + jobId;
-            VfcResponse responseGet = null;
-
-            while (attemptsLeft-- > 0) {
-                NetLoggerUtil.getNetworkLogger().info("[OUT|{}|{}|]", "VFC", urlGet);
-                Pair httpDetailsGet = restManager.get(urlGet, username, password, headers);
-                responseGet = Serialization.gsonPretty.fromJson(httpDetailsGet.second, VfcResponse.class);
-                NetLoggerUtil.log(EventType.IN, CommInfrastructure.REST, vfcUrl, httpDetailsGet.second);
-                responseGet.setRequestId(vfcRequest.getRequestId().toString());
-                body = Serialization.gsonPretty.toJson(responseGet);
-                logger.debug("Response to VFC Heal get:");
-                logger.debug(body);
-
-                String responseStatus = responseGet.getResponseDescriptor().getStatus();
-                if (httpDetailsGet.first == 200
-                        && ("finished".equalsIgnoreCase(responseStatus) || "error".equalsIgnoreCase(responseStatus))) {
-                    logger.debug("VFC Heal Status {}", responseGet.getResponseDescriptor().getStatus());
-                    this.callback.onResponse(responseGet);
-                    break;
-                }
-                Thread.sleep(20000);
-            }
-            boolean isTimeout = (attemptsLeft <= 0) && (responseGet != null)
-                            && (responseGet.getResponseDescriptor() != null);
-            isTimeout = isTimeout && (responseGet.getResponseDescriptor().getStatus() != null)
-                            && (!responseGet.getResponseDescriptor().getStatus().isEmpty());
-            if (isTimeout) {
-                logger.debug("VFC timeout. Status: ({})", responseGet.getResponseDescriptor().getStatus());
-                this.callback.onResponse(responseGet);
-            }
+            handleVfcResponse(headers, httpDetails, vfcUrl);
         } catch (JsonSyntaxException e) {
             logger.error("Failed to deserialize into VfcResponse {}", e.getLocalizedMessage(), e);
         } catch (InterruptedException e) {
@@ -174,6 +136,57 @@ public final class VfcManager implements Runnable {
         }
     }
 
+    /**
+     * Handle a VFC response message.
+     *
+     * @param headers the headers in the response
+     * @param httpDetails the HTTP details in the response
+     * @param vfcUrl the response URL
+     * @throws InterruptedException on errors in the response
+     */
+    private void handleVfcResponse(Map headers, Pair httpDetails, String vfcUrl)
+            throws InterruptedException {
+        VfcResponse response = Serialization.gsonPretty.fromJson(httpDetails.second, VfcResponse.class);
+        NetLoggerUtil.log(EventType.IN, CommInfrastructure.REST, vfcUrl, httpDetails.second);
+        String body = Serialization.gsonPretty.toJson(response);
+        logger.debug("Response to VFC Heal post:");
+        logger.debug(body);
+
+        String jobId = response.getJobId();
+        int attemptsLeft = 20;
+
+        String urlGet = vfcUrlBase + "/jobs/" + jobId;
+        VfcResponse responseGet = null;
+
+        while (attemptsLeft-- > 0) {
+            NetLoggerUtil.getNetworkLogger().info("[OUT|{}|{}|]", "VFC", urlGet);
+            Pair httpDetailsGet = restManager.get(urlGet, username, password, headers);
+            responseGet = Serialization.gsonPretty.fromJson(httpDetailsGet.second, VfcResponse.class);
+            NetLoggerUtil.log(EventType.IN, CommInfrastructure.REST, vfcUrl, httpDetailsGet.second);
+            responseGet.setRequestId(vfcRequest.getRequestId().toString());
+            body = Serialization.gsonPretty.toJson(responseGet);
+            logger.debug("Response to VFC Heal get:");
+            logger.debug(body);
+
+            String responseStatus = responseGet.getResponseDescriptor().getStatus();
+            if (httpDetailsGet.first == 200
+                    && ("finished".equalsIgnoreCase(responseStatus) || "error".equalsIgnoreCase(responseStatus))) {
+                logger.debug("VFC Heal Status {}", responseGet.getResponseDescriptor().getStatus());
+                this.callback.onResponse(responseGet);
+                return;
+            }
+            Thread.sleep(20000);
+        }
+        boolean isTimeout = (attemptsLeft <= 0) && (responseGet != null)
+                        && (responseGet.getResponseDescriptor() != null);
+        isTimeout = isTimeout && (responseGet.getResponseDescriptor().getStatus() != null)
+                        && (!responseGet.getResponseDescriptor().getStatus().isEmpty());
+        if (isTimeout) {
+            logger.debug("VFC timeout. Status: ({})", responseGet.getResponseDescriptor().getStatus());
+            this.callback.onResponse(responseGet);
+        }
+    }
+
     /**
      * Protected setter for rest manager to allow mocked rest manager to be used for testing.
      *
diff --git a/models-interactions/model-simulators/src/main/java/org/onap/policy/simulators/AaiSimulatorJaxRs.java b/models-interactions/model-simulators/src/main/java/org/onap/policy/simulators/AaiSimulatorJaxRs.java
index 756253c47..793babf24 100644
--- a/models-interactions/model-simulators/src/main/java/org/onap/policy/simulators/AaiSimulatorJaxRs.java
+++ b/models-interactions/model-simulators/src/main/java/org/onap/policy/simulators/AaiSimulatorJaxRs.java
@@ -157,8 +157,8 @@ public class AaiSimulatorJaxRs {
     @Produces("application/json")
     public String getByVnfName(@QueryParam("vnf-name") final String vnfName) {
         if (GETFAIL.equals(vnfName)) {
-            return "{\"requestError\":{\"serviceException\":{\"messageId\":\"SVC3001\",\"text\":\"Resource not found"
-                    + " for %1 using id %2 (msg=%3) (ec=%4)\",\"variables\":[\"GET\",\"network/generic-vnfs/"
+            return "{\"requestError\":{\"serviceException\":{\"messageId\":\"SVC3001\",\"text\":\"Resource not"
+                    + " found for %1 using id %2 (msg=%3) (ec=%4)\",\"variables\":[\"GET\",\"network/generic-vnfs/"
                     + "generic-vnf\",\"Node Not Found:No Node of type generic-vnf found at network/generic-vnfs"
                     + "/generic-vnf\",\"ERR.5.4.6114\"]}}}";
         }
diff --git a/models-interactions/model-simulators/src/main/java/org/onap/policy/simulators/Util.java b/models-interactions/model-simulators/src/main/java/org/onap/policy/simulators/Util.java
index 6c1a05753..f21fb6f38 100644
--- a/models-interactions/model-simulators/src/main/java/org/onap/policy/simulators/Util.java
+++ b/models-interactions/model-simulators/src/main/java/org/onap/policy/simulators/Util.java
@@ -21,12 +21,12 @@
 
 package org.onap.policy.simulators;
 
-import java.io.IOException;
 import java.util.Properties;
 import org.onap.policy.common.endpoints.http.server.HttpServletServer;
 import org.onap.policy.common.endpoints.http.server.HttpServletServerFactoryInstance;
 import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
 import org.onap.policy.common.gson.GsonMessageBodyHandler;
+import org.onap.policy.common.parameters.ParameterRuntimeException;
 import org.onap.policy.common.utils.coder.CoderException;
 import org.onap.policy.common.utils.coder.StandardCoder;
 import org.onap.policy.common.utils.network.NetworkUtil;
@@ -49,11 +49,12 @@ public class Util {
     public static final int SDNCSIM_SERVER_PORT = 6670;
     public static final int DMAAPSIM_SERVER_PORT = 3904;
 
+    private static final String CANNOT_PROCESS_PARAMETERS = "cannot parse parameters ";
     private static final String CANNOT_CONNECT = "cannot connect to port ";
     private static final String LOCALHOST = "localhost";
 
     private Util() {
-        // Prevent instantiation of thic class
+        // Prevent instantiation of this class
     }
 
     /**
@@ -61,11 +62,10 @@ public class Util {
      *
      * @return the simulator
      * @throws InterruptedException if a thread is interrupted
-     * @throws IOException if an IO errror occurs
      */
-    public static HttpServletServer buildAaiSim() throws InterruptedException, IOException {
+    public static HttpServletServer buildAaiSim() throws InterruptedException {
         final HttpServletServer testServer = HttpServletServerFactoryInstance.getServerFactory()
-                        .build(AAISIM_SERVER_NAME, LOCALHOST, AAISIM_SERVER_PORT, "/", false, true);
+                .build(AAISIM_SERVER_NAME, LOCALHOST, AAISIM_SERVER_PORT, "/", false, true);
         testServer.addServletClass("/*", AaiSimulatorJaxRs.class.getName());
         testServer.waitedStart(5000);
         if (!NetworkUtil.isTcpPortOpen(LOCALHOST, testServer.getPort(), 5, 10000L)) {
@@ -79,11 +79,10 @@ public class Util {
      *
      * @return the simulator
      * @throws InterruptedException if a thread is interrupted
-     * @throws IOException if an IO errror occurs
      */
-    public static HttpServletServer buildSdncSim() throws InterruptedException, IOException {
+    public static HttpServletServer buildSdncSim() throws InterruptedException {
         final HttpServletServer testServer = HttpServletServerFactoryInstance.getServerFactory()
-                        .build(SDNCSIM_SERVER_NAME, LOCALHOST, SDNCSIM_SERVER_PORT, "/", false, true);
+                .build(SDNCSIM_SERVER_NAME, LOCALHOST, SDNCSIM_SERVER_PORT, "/", false, true);
         testServer.addServletClass("/*", SdncSimulatorJaxRs.class.getName());
         testServer.waitedStart(5000);
         if (!NetworkUtil.isTcpPortOpen(LOCALHOST, testServer.getPort(), 5, 10000L)) {
@@ -98,11 +97,10 @@ public class Util {
      *
      * @return the simulator
      * @throws InterruptedException if a thread is interrupted
-     * @throws IOException if an IO errror occurs
      */
-    public static HttpServletServer buildSoSim() throws InterruptedException, IOException {
+    public static HttpServletServer buildSoSim() throws InterruptedException {
         final HttpServletServer testServer = HttpServletServerFactoryInstance.getServerFactory()
-                        .build(SOSIM_SERVER_NAME, LOCALHOST, SOSIM_SERVER_PORT, "/", false, true);
+                .build(SOSIM_SERVER_NAME, LOCALHOST, SOSIM_SERVER_PORT, "/", false, true);
         testServer.addServletClass("/*", SoSimulatorJaxRs.class.getName());
         testServer.waitedStart(5000);
         if (!NetworkUtil.isTcpPortOpen(LOCALHOST, testServer.getPort(), 5, 10000L)) {
@@ -116,11 +114,10 @@ public class Util {
      *
      * @return the simulator
      * @throws InterruptedException if a thread is interrupted
-     * @throws IOException if an IO errror occurs
      */
-    public static HttpServletServer buildVfcSim() throws InterruptedException, IOException {
+    public static HttpServletServer buildVfcSim() throws InterruptedException {
         final HttpServletServer testServer = HttpServletServerFactoryInstance.getServerFactory()
-                        .build(VFCSIM_SERVER_NAME, LOCALHOST, VFCSIM_SERVER_PORT, "/", false, true);
+                .build(VFCSIM_SERVER_NAME, LOCALHOST, VFCSIM_SERVER_PORT, "/", false, true);
         testServer.addServletClass("/*", VfcSimulatorJaxRs.class.getName());
         testServer.waitedStart(5000);
         if (!NetworkUtil.isTcpPortOpen(LOCALHOST, testServer.getPort(), 5, 10000L)) {
@@ -134,11 +131,10 @@ public class Util {
      *
      * @return the simulator
      * @throws InterruptedException if a thread is interrupted
-     * @throws IOException if an IO errror occurs
      */
-    public static HttpServletServer buildGuardSim() throws InterruptedException, IOException {
+    public static HttpServletServer buildGuardSim() throws InterruptedException {
         HttpServletServer testServer = HttpServletServerFactoryInstance.getServerFactory().build(GUARDSIM_SERVER_NAME,
-                        LOCALHOST, GUARDSIM_SERVER_PORT, "/", false, true);
+                LOCALHOST, GUARDSIM_SERVER_PORT, "/", false, true);
         testServer.setSerializationProvider(GsonMessageBodyHandler.class.getName());
         testServer.addServletClass("/*", GuardSimulatorJaxRs.class.getName());
         testServer.waitedStart(5000);
@@ -153,21 +149,25 @@ public class Util {
      *
      * @return the simulator
      * @throws InterruptedException if a thread is interrupted
-     * @throws IOException if an IO errror occurs
-     * @throws CoderException if the server parameters cannot be loaded
      */
-    public static HttpServletServer buildDmaapSim() throws InterruptedException, IOException, CoderException {
+    public static HttpServletServer buildDmaapSim() throws InterruptedException {
         String json = ResourceUtils.getResourceAsString("org/onap/policy/simulators/dmaap/DmaapParameters.json");
-        DmaapSimParameterGroup params = new StandardCoder().decode(json, DmaapSimParameterGroup.class);
+        DmaapSimParameterGroup params = null;
+        try {
+            params = new StandardCoder().decode(json, DmaapSimParameterGroup.class);
+        } catch (CoderException ce) {
+            throw new ParameterRuntimeException(
+                    CANNOT_PROCESS_PARAMETERS + "org/onap/policy/simulators/dmaap/DmaapParameters.json", ce);
+        }
 
         DmaapSimProvider.setInstance(new DmaapSimProvider(params));
 
         Properties props = DmaapSimRestServer.getServerProperties(params.getRestServerParameters());
 
         final String svcpfx = PolicyEndPointProperties.PROPERTY_HTTP_SERVER_SERVICES + "."
-                        + params.getRestServerParameters().getName();
+                + params.getRestServerParameters().getName();
         props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_HTTP_PORT_SUFFIX,
-                        Integer.toString(DMAAPSIM_SERVER_PORT));
+                Integer.toString(DMAAPSIM_SERVER_PORT));
         props.setProperty(svcpfx + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX, "true");
 
         HttpServletServer testServer = HttpServletServerFactoryInstance.getServerFactory().build(props).get(0);
diff --git a/models-pap/src/main/java/org/onap/policy/models/pap/concepts/PolicyNotification.java b/models-pap/src/main/java/org/onap/policy/models/pap/concepts/PolicyNotification.java
new file mode 100644
index 000000000..a9c22c463
--- /dev/null
+++ b/models-pap/src/main/java/org/onap/policy/models/pap/concepts/PolicyNotification.java
@@ -0,0 +1,48 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.models.pap.concepts;
+
+import com.google.gson.annotations.SerializedName;
+import java.util.ArrayList;
+import java.util.List;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+public class PolicyNotification {
+
+    /**
+     * Status of policies that are being added to PDPs.
+     */
+    @SerializedName("added-policies")
+    private List added = new ArrayList<>();
+
+    /**
+     * Status of policies that are being deleted from PDPs.
+     */
+    @SerializedName("deleted-policies")
+    private List deleted = new ArrayList<>();
+
+
+    public PolicyNotification(List added, List deleted) {
+        this.added = added;
+        this.deleted = deleted;
+    }
+}
diff --git a/models-pap/src/main/java/org/onap/policy/models/pap/concepts/PolicyStatus.java b/models-pap/src/main/java/org/onap/policy/models/pap/concepts/PolicyStatus.java
new file mode 100644
index 000000000..56d6c6909
--- /dev/null
+++ b/models-pap/src/main/java/org/onap/policy/models/pap/concepts/PolicyStatus.java
@@ -0,0 +1,60 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.models.pap.concepts;
+
+import com.google.gson.annotations.SerializedName;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyIdentifier;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyTypeIdentifier;
+
+@Data
+@NoArgsConstructor
+public class PolicyStatus {
+
+    @SerializedName("policy-type")
+    private ToscaPolicyTypeIdentifier policyType;
+
+    @SerializedName("policy")
+    private ToscaPolicyIdentifier policy;
+
+    /**
+     * Number of PDPs that have successfully added/deleted the policy.
+     */
+    @SerializedName("success-count")
+    private int successCount = 0;
+
+    /**
+     * Number of PDPs that were unable to add/delete the policy.
+     */
+    @SerializedName("failure-count")
+    private int failureCount = 0;
+
+    /**
+     * Number of PDPs that have not completed the add/delete operation.
+     */
+    @SerializedName("incomplete-count")
+    private int incompleteCount = 0;
+
+
+    public PolicyStatus(ToscaPolicyTypeIdentifier policyType, ToscaPolicyIdentifier policy) {
+        this.policyType = policyType;
+        this.policy = policy;
+    }
+}
diff --git a/models-pap/src/test/java/org/onap/policy/models/pap/concepts/PolicyNotificationTest.java b/models-pap/src/test/java/org/onap/policy/models/pap/concepts/PolicyNotificationTest.java
new file mode 100644
index 000000000..3c7b0cab6
--- /dev/null
+++ b/models-pap/src/test/java/org/onap/policy/models/pap/concepts/PolicyNotificationTest.java
@@ -0,0 +1,60 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.models.pap.concepts;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+
+import java.util.Arrays;
+import java.util.List;
+import org.junit.Test;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.common.utils.coder.StandardCoder;
+
+/**
+ * This only tests the methods that aren't already tested via TestModels.
+ */
+public class PolicyNotificationTest {
+
+    @Test
+    public void test() throws CoderException {
+        PolicyStatus statusAdd1 = new PolicyStatus();
+        statusAdd1.setSuccessCount(10);
+        PolicyStatus statusAdd2 = new PolicyStatus();
+        statusAdd2.setFailureCount(20);
+        List add = Arrays.asList(statusAdd1, statusAdd2);
+
+        PolicyStatus statusDel1 = new PolicyStatus();
+        statusDel1.setIncompleteCount(30);
+        PolicyStatus statusDel2 = new PolicyStatus();
+        List del = Arrays.asList(statusDel1, statusDel2);
+
+        // test constructor with arguments
+        PolicyNotification notify = new PolicyNotification(add, del);
+        assertSame(add, notify.getAdded());
+        assertSame(del, notify.getDeleted());
+
+        // encode & decode
+        StandardCoder coder = new StandardCoder();
+        PolicyNotification notify2 = coder.decode(coder.encode(notify), PolicyNotification.class);
+
+        // test equals() method (and verify encode/decode worked)
+        assertEquals(notify, notify2);
+    }
+}
diff --git a/models-pap/src/test/java/org/onap/policy/models/pap/concepts/PolicyStatusTest.java b/models-pap/src/test/java/org/onap/policy/models/pap/concepts/PolicyStatusTest.java
new file mode 100644
index 000000000..0a00e7b60
--- /dev/null
+++ b/models-pap/src/test/java/org/onap/policy/models/pap/concepts/PolicyStatusTest.java
@@ -0,0 +1,61 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.models.pap.concepts;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+
+import org.junit.Test;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.common.utils.coder.StandardCoder;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyIdentifier;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyTypeIdentifier;
+
+/**
+ * This only tests the methods that aren't already tested via TestModels.
+ */
+public class PolicyStatusTest {
+
+    @Test
+    public void test() throws CoderException {
+        ToscaPolicyTypeIdentifier type = new ToscaPolicyTypeIdentifier("my-type", "3.2.1");
+        ToscaPolicyIdentifier policy = new ToscaPolicyIdentifier("my-name", "1.2.3");
+
+        // test constructor with arguments
+        PolicyStatus status = new PolicyStatus(type, policy);
+        assertSame(type, status.getPolicyType());
+        assertSame(policy, status.getPolicy());
+
+        assertEquals(0, status.getSuccessCount());
+        assertEquals(0, status.getFailureCount());
+        assertEquals(0, status.getIncompleteCount());
+
+        // change values
+        status.setFailureCount(10);
+        status.setIncompleteCount(20);
+        status.setSuccessCount(30);
+
+        // encode & decode
+        StandardCoder coder = new StandardCoder();
+        PolicyStatus status2 = coder.decode(coder.encode(status), PolicyStatus.class);
+
+        // test equals() method (and verify encode/decode worked)
+        assertEquals(status, status2);
+    }
+}
diff --git a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/e2e/EndToEndTest.java b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/e2e/EndToEndTest.java
index 8c35de64e..50c319837 100644
--- a/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/e2e/EndToEndTest.java
+++ b/models-sim/models-sim-dmaap/src/test/java/org/onap/policy/sim/dmaap/e2e/EndToEndTest.java
@@ -32,7 +32,7 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
-import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -50,27 +50,24 @@ import org.onap.policy.models.sim.dmaap.rest.CommonRestServer;
  */
 public class EndToEndTest extends CommonRestServer {
     private static final int MAX_WAIT_SEC = 5;
-    private static final String ORIG_TOPIC = "MY-TOPIC";
-    private static final String ORIG_TOPIC2 = "MY-TOPIC-B";
+    private static final String TOPIC = "MY-TOPIC";
+    private static final String TOPIC2 = "MY-TOPIC-B";
     private static final int MAX_MSG = 200;
 
-    private static int ntests = 0;
-    private static String topicJson;
-
-    private TopicParameterGroup topicConfig;
-
-    private String topic = "MY-TOPIC";
-    private String topic2 = "MY-TOPIC-B";
-
     /**
      * Messages from the topic are placed here by the endpoint.
      */
-    private BlockingQueue queue;
+    private static BlockingQueue queue;
 
     /**
      * Messages from topic 2 are placed here by the endpoint.
      */
-    private BlockingQueue queue2;
+    private static BlockingQueue queue2;
+
+    /**
+     * Topic configuration parameters.
+     */
+    private static TopicParameterGroup topicConfig;
 
     /**
      * Starts the rest server.
@@ -83,10 +80,34 @@ public class EndToEndTest extends CommonRestServer {
 
         CommonRestServer.setUpBeforeClass();
 
-        topicJson = new String(
+        queue = new LinkedBlockingQueue<>();
+        queue2 = new LinkedBlockingQueue<>();
+
+        String json = new String(
                         Files.readAllBytes(new File("src/test/resources/parameters/TopicParameters.json").toPath()),
                         StandardCharsets.UTF_8);
-        topicJson = topicJson.replace("${port}", String.valueOf(getPort()));
+        json = json.replace("${port}", String.valueOf(getPort()));
+
+        topicConfig = new StandardCoder().decode(json, TopicParameterGroup.class);
+
+        TopicEndpointManager.getManager().addTopics(topicConfig);
+        TopicEndpointManager.getManager().start();
+
+        TopicEndpointManager.getManager().getDmaapTopicSource(TOPIC)
+                        .register((infra, topic, event) -> queue.add(event));
+        TopicEndpointManager.getManager().getDmaapTopicSource(TOPIC2)
+                        .register((infra, topic, event) -> queue2.add(event));
+    }
+
+    /**
+     * Stops the topics and clears the queues.
+     */
+    @AfterClass
+    public static void tearDownAfterClass() {
+        TopicEndpointManager.getManager().shutdown();
+
+        queue = null;
+        queue2 = null;
     }
 
     /**
@@ -97,42 +118,15 @@ public class EndToEndTest extends CommonRestServer {
     @Before
     @Override
     public void setUp() throws CoderException {
-        queue = new LinkedBlockingQueue<>();
-        queue2 = new LinkedBlockingQueue<>();
-
-        /*
-         * change topic names for each test so any listeners that may still exist will not
-         * grab new messages
-         */
-
-        ++ntests;
-        topic = "my-topic-" + ntests;
-        topic2 = "my-topic-b" + ntests;
-
-        String json = topicJson.replace(ORIG_TOPIC2, topic2).replace(ORIG_TOPIC, topic);
-
-        topicConfig = new StandardCoder().decode(json, TopicParameterGroup.class);
-
-        TopicEndpointManager.getManager().addTopics(topicConfig);
-        TopicEndpointManager.getManager().start();
-    }
-
-    @After
-    public void tearDown() {
-        TopicEndpointManager.getManager().shutdown();
+        queue.clear();
+        queue2.clear();
     }
 
     @Test
     public void testWithTopicEndpointAtEachEnd() throws InterruptedException {
-        // register listeners to add events to appropriate queue
-        TopicEndpointManager.getManager().getDmaapTopicSource(topic)
-                        .register((infra, topic, event) -> queue.add(event));
-        TopicEndpointManager.getManager().getDmaapTopicSource(topic2)
-                        .register((infra, topic, event) -> queue2.add(event));
-
         // publish events
-        TopicSink sink = TopicEndpointManager.getManager().getDmaapTopicSink(topic);
-        TopicSink sink2 = TopicEndpointManager.getManager().getDmaapTopicSink(topic2);
+        TopicSink sink = TopicEndpointManager.getManager().getDmaapTopicSink(TOPIC);
+        TopicSink sink2 = TopicEndpointManager.getManager().getDmaapTopicSink(TOPIC2);
         for (int x = 0; x < MAX_MSG; ++x) {
             sink.send("hello-" + x);
             sink2.send("world-" + x);
@@ -147,8 +141,10 @@ public class EndToEndTest extends CommonRestServer {
 
     @Test
     public void testCambriaFormat() throws Exception {
+        // @formatter:off
         test("testCambriaFormat", "application/cambria",
             (wtr, messages) -> messages.forEach(msg -> wtr.write("0." + msg.length() + "." + msg + "\n")));
+        // @formatter:on
     }
 
     @Test
@@ -175,9 +171,6 @@ public class EndToEndTest extends CommonRestServer {
         String msg1 = "{'abc':10.0}".replace('\'', '"');
         String msg2 = "{'def':20.0}".replace('\'', '"');
 
-        TopicEndpointManager.getManager().getDmaapTopicSource(topic)
-                        .register((infra, topic, event) -> queue.add(event));
-
         TopicParameters sinkcfg = topicConfig.getTopicSinks().get(0);
         URL url = new URL(httpPrefix + "events/" + sinkcfg.getTopic());
 
diff --git a/models-sim/policy-models-sim-pdp/src/main/java/org/onap/policy/models/sim/pdp/handler/PdpUpdateMessageHandler.java b/models-sim/policy-models-sim-pdp/src/main/java/org/onap/policy/models/sim/pdp/handler/PdpUpdateMessageHandler.java
index 4f33d079f..5b77f4704 100644
--- a/models-sim/policy-models-sim-pdp/src/main/java/org/onap/policy/models/sim/pdp/handler/PdpUpdateMessageHandler.java
+++ b/models-sim/policy-models-sim-pdp/src/main/java/org/onap/policy/models/sim/pdp/handler/PdpUpdateMessageHandler.java
@@ -91,12 +91,18 @@ public class PdpUpdateMessageHandler {
      * @return boolean flag which tells if the information is same or not
      */
     private boolean checkIfAlreadyHandled(final PdpUpdate pdpUpdateMsg, final PdpStatus pdpStatusContext) {
-        return null != pdpStatusContext.getPdpGroup()
-                && pdpStatusContext.getPdpGroup().equals(pdpUpdateMsg.getPdpGroup())
-                && null != pdpStatusContext.getPdpSubgroup()
-                && pdpStatusContext.getPdpSubgroup().equals(pdpUpdateMsg.getPdpSubgroup())
-                && null != pdpStatusContext.getPolicies() && new PdpMessageHandler()
-                        .getToscaPolicyIdentifiers(pdpUpdateMsg.getPolicies()).equals(pdpStatusContext.getPolicies());
+        if (pdpStatusContext.getPdpGroup() == null
+                || !pdpStatusContext.getPdpGroup().equals(pdpUpdateMsg.getPdpGroup())) {
+            return false;
+        }
+
+        if (pdpStatusContext.getPdpSubgroup() == null
+                || !pdpStatusContext.getPdpSubgroup().equals(pdpUpdateMsg.getPdpSubgroup())) {
+            return false;
+        }
+
+        return null != pdpStatusContext.getPolicies() && new PdpMessageHandler()
+                .getToscaPolicyIdentifiers(pdpUpdateMsg.getPolicies()).equals(pdpStatusContext.getPolicies());
     }
 
     /**