Add support for unique replica id generation in the participants

Issue-ID: POLICY-5032
Change-Id: I9c36b87d1f03f03089d2c62308e0975e24f6e99a
Signed-off-by: FrancescoFioraEst <francesco.fiora@est.tech>

5 files changed, 39 insertions, 17 deletions

diff --git a/models/src/main/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantAckMessage.java b/models/src/main/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantAckMessage.java
index fa3e3d08e..7ea3310be 100644
--- a/models/src/main/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantAckMessage.java
+++ b/models/src/main/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantAckMessage.java
@@ -57,6 +57,8 @@ public class ParticipantAckMessage {
      */
     private UUID participantId;
 
+    private UUID replicaId;
+
     /**
      * Participant State, or {@code null} for messages from participants.
      */
@@ -82,7 +84,9 @@ public class ParticipantAckMessage {
         this.stateChangeResult = source.stateChangeResult;
         this.message = source.message;
         this.messageType = source.messageType;
+        this.compositionId = source.compositionId;
         this.participantId = source.participantId;
+        this.replicaId = source.replicaId;
         this.state = source.state;
     }
 
@@ -90,15 +94,17 @@ public class ParticipantAckMessage {
      * Determines if this message applies to this participant type.
      *
      * @param participantId id of the participant to match against
+     * @param replicaId id of the participant to match against
      * @return {@code true} if this message applies to this participant, {@code false} otherwise
      */
-    public boolean appliesTo(@NonNull final UUID participantId) {
+    public boolean appliesTo(@NonNull final UUID participantId, @NonNull final UUID replicaId) {
         // Broadcast message to all participants
-        if (this.participantId == null) {
+        if ((this.participantId == null)
+                || (participantId.equals(this.participantId) && this.replicaId == null)) {
             return true;
         }
 
         // Targeted message at this specific participant
-        return participantId.equals(this.participantId);
+        return replicaId.equals(this.replicaId);
     }
 }
diff --git a/models/src/main/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantMessage.java b/models/src/main/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantMessage.java
index 304db8eb5..f8aea947b 100644
--- a/models/src/main/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantMessage.java
+++ b/models/src/main/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantMessage.java
@@ -51,6 +51,8 @@ public class ParticipantMessage {
      */
     private UUID participantId;
 
+    private UUID replicaId;
+
     /**
      * Automation Composition ID, or {@code null} for messages to participants.
      */
@@ -75,6 +77,7 @@ public class ParticipantMessage {
     public ParticipantMessage(final ParticipantMessage source) {
         this.messageType = source.messageType;
         this.participantId = source.participantId;
+        this.replicaId = source.replicaId;
         this.automationCompositionId = source.automationCompositionId;
         this.compositionId = source.compositionId;
     }
@@ -83,15 +86,17 @@ public class ParticipantMessage {
      * Determines if this message applies to this participant type.
      *
      * @param participantId id of the participant to match against
+     * @param replicaId id of the participant to match against
      * @return {@code true} if this message applies to this participant, {@code false} otherwise
      */
-    public boolean appliesTo(@NonNull final UUID participantId) {
+    public boolean appliesTo(@NonNull final UUID participantId, @NonNull final UUID replicaId) {
         // Broadcast message to all participants
-        if (this.participantId == null) {
+        if ((this.participantId == null)
+                || (participantId.equals(this.participantId) && this.replicaId == null)) {
             return true;
         }
 
         // Targeted message at this specific participant
-        return participantId.equals(this.participantId);
+        return replicaId.equals(this.replicaId);
     }
 }
diff --git a/models/src/test/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantAckMessageTest.java b/models/src/test/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantAckMessageTest.java
index b6af01c44..72e4efb49 100644
--- a/models/src/test/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantAckMessageTest.java
+++ b/models/src/test/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantAckMessageTest.java
@@ -57,7 +57,8 @@ class ParticipantAckMessageTest {
     @Test
     void testAppliesTo_NullParticipantId() {
         message = makeMessage();
-        assertThatThrownBy(() -> message.appliesTo(null)).isInstanceOf(NullPointerException.class);
+        assertThatThrownBy(() -> message.appliesTo(UUID.randomUUID(), null)).isInstanceOf(NullPointerException.class);
+        assertThatThrownBy(() -> message.appliesTo(null, UUID.randomUUID())).isInstanceOf(NullPointerException.class);
     }
 
     @Test
@@ -65,8 +66,8 @@ class ParticipantAckMessageTest {
         message = makeMessage();
 
         // ParticipantId matches
-        assertTrue(message.appliesTo(CommonTestData.getParticipantId()));
-        assertFalse(message.appliesTo(CommonTestData.getRndParticipantId()));
+        assertTrue(message.appliesTo(CommonTestData.getParticipantId(), CommonTestData.getReplicaId()));
+        assertFalse(message.appliesTo(CommonTestData.getRndParticipantId(), CommonTestData.getReplicaId()));
     }
 
     @Test
@@ -74,8 +75,8 @@ class ParticipantAckMessageTest {
         message = makeMessage();
 
         // ParticipantId does not match
-        assertFalse(message.appliesTo(CommonTestData.getRndParticipantId()));
-        assertTrue(message.appliesTo(CommonTestData.getParticipantId()));
+        assertFalse(message.appliesTo(CommonTestData.getRndParticipantId(), CommonTestData.getReplicaId()));
+        assertTrue(message.appliesTo(CommonTestData.getParticipantId(), CommonTestData.getReplicaId()));
     }
 
     private ParticipantAckMessage makeMessage() {
diff --git a/models/src/test/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantMessageTest.java b/models/src/test/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantMessageTest.java
index 541d8ef31..db31d0f01 100644
--- a/models/src/test/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantMessageTest.java
+++ b/models/src/test/java/org/onap/policy/clamp/models/acm/messages/kafka/participant/ParticipantMessageTest.java
@@ -61,7 +61,8 @@ class ParticipantMessageTest {
     void testAppliesTo_NullParticipantId() {
         message = makeMessage();
 
-        assertThatThrownBy(() -> message.appliesTo(null)).isInstanceOf(NullPointerException.class);
+        assertThatThrownBy(() -> message.appliesTo(UUID.randomUUID(), null)).isInstanceOf(NullPointerException.class);
+        assertThatThrownBy(() -> message.appliesTo(null, UUID.randomUUID())).isInstanceOf(NullPointerException.class);
     }
 
     @Test
@@ -69,15 +70,15 @@ class ParticipantMessageTest {
         message = makeMessage();
 
         // ParticipantId matches
-        assertTrue(message.appliesTo(CommonTestData.getParticipantId()));
-        assertFalse(message.appliesTo(CommonTestData.getRndParticipantId()));
+        assertTrue(message.appliesTo(CommonTestData.getParticipantId(), CommonTestData.getReplicaId()));
+        assertFalse(message.appliesTo(CommonTestData.getRndParticipantId(), CommonTestData.getReplicaId()));
     }
 
     @Test
     void testAppliesTo_ParticipantIdNoMatch() {
         message = makeMessage();
-        assertFalse(message.appliesTo(CommonTestData.getRndParticipantId()));
-        assertTrue(message.appliesTo(CommonTestData.getParticipantId()));
+        assertFalse(message.appliesTo(CommonTestData.getRndParticipantId(), CommonTestData.getReplicaId()));
+        assertTrue(message.appliesTo(CommonTestData.getParticipantId(), CommonTestData.getReplicaId()));
     }
 
     private ParticipantMessage makeMessage() {
diff --git a/models/src/test/java/org/onap/policy/clamp/models/acm/utils/CommonTestData.java b/models/src/test/java/org/onap/policy/clamp/models/acm/utils/CommonTestData.java
index 131c8eefd..b8075c3ef 100644
--- a/models/src/test/java/org/onap/policy/clamp/models/acm/utils/CommonTestData.java
+++ b/models/src/test/java/org/onap/policy/clamp/models/acm/utils/CommonTestData.java
@@ -38,9 +38,9 @@ import org.onap.policy.models.tosca.authorative.concepts.ToscaServiceTemplate;
 public class CommonTestData {
 
     public static final UUID PARTCICIPANT_ID = UUID.randomUUID();
+    public static final UUID REPLICA_ID = UUID.randomUUID();
     private static final StandardYamlCoder YAML_TRANSLATOR = new StandardYamlCoder();
 
-
     /**
      * Returns participantId for test cases.
      *
@@ -51,6 +51,15 @@ public class CommonTestData {
     }
 
     /**
+     * Returns participantId for test cases.
+     *
+     * @return participant Id
+     */
+    public static UUID getReplicaId() {
+        return REPLICA_ID;
+    }
+
+    /**
      * Returns participantId for test Jpa cases.
      *
      * @return participant Id