Do not publish SubjectDeletedAnnouncement for updated subjects.

Signed-off-by: Yufei Cai <yufei.cai@bosch.io>

policies/service/src/main/java/org/eclipse/ditto/policies/service/persistence/actors/announcements/PolicyAnnouncementManager.java

@@ -27,12 +27,14 @@
 import org.eclipse.ditto.policies.model.PolicyId;
 import org.eclipse.ditto.policies.model.PolicyLifecycle;
 import org.eclipse.ditto.policies.model.Subject;
+import org.eclipse.ditto.policies.model.SubjectId;
 import org.eclipse.ditto.policies.model.Subjects;
 import org.eclipse.ditto.policies.model.signals.announcements.PolicyAnnouncement;
 import org.eclipse.ditto.policies.service.common.config.PolicyAnnouncementConfig;
 
 import akka.actor.AbstractActor;
 import akka.actor.ActorRef;
+import akka.actor.PoisonPill;
 import akka.actor.Props;
 import akka.actor.Terminated;
 import akka.japi.pf.ReceiveBuilder;
@@ -48,6 +50,7 @@ public final class PolicyAnnouncementManager extends AbstractActor {
     private final DistributedPub<PolicyAnnouncement<?>> policyAnnouncementPub;
     private final Map<Subject, ActorRef> subjectExpiryActors;
     private final Map<ActorRef, Subject> activeSubjects;
+    private final Map<SubjectId, Integer> activeSubjectIds;
     private final ActorRef commandForwarder;
     private final PolicyAnnouncementConfig config;
 
@@ -61,6 +64,7 @@ private PolicyAnnouncementManager(final PolicyId policyId,
         this.commandForwarder = commandForwarder;
         this.subjectExpiryActors = new HashMap<>();
         this.activeSubjects = new HashMap<>();
+        this.activeSubjectIds = new HashMap<>();
         this.config = config;
     }
 
@@ -99,6 +103,7 @@ private void onPolicyModified(final Policy policy) {
             startChild(newSubject);
         }
         for (final var deletedSubject : deletedSubjects) {
+            // precondition: child actors have started for new subjects so that modified subjects can be recognized
             sendSubjectDeleted(deletedSubject);
         }
     }
@@ -112,13 +117,15 @@ private void startChild(final Subject subject) {
         getContext().watch(child);
         subjectExpiryActors.put(subject, child);
         activeSubjects.put(child, subject);
+        addActiveSubjectId(subject);
     }
 
     private void onChildTerminated(final Terminated terminated) {
         final var terminatedActor = terminated.actor();
         final var removedSubject = activeSubjects.remove(terminatedActor);
         if (removedSubject != null) {
             final var child = subjectExpiryActors.remove(removedSubject);
+            removeActiveSubjectId(removedSubject);
             log.debug("OnChildTerminated: Removed terminated child <{}>", child);
         } else {
             log.debug("OnChildTerminated: Child not found: <{}>", terminatedActor);
@@ -128,12 +135,30 @@ private void onChildTerminated(final Terminated terminated) {
     private void sendSubjectDeleted(final Subject subject) {
         final var child = subjectExpiryActors.get(subject);
         if (child != null) {
-            child.tell(SubjectExpiryActor.Message.SUBJECT_DELETED, ActorRef.noSender());
+            notifyChildOfSubjectDeletion(subject, child);
         } else {
             log.error("Attempting to notify nonexistent child for deleted subject <{}>", subject);
         }
     }
 
+    /**
+     * Notify a child whose subject was deleted from the policy.
+     *
+     * @param subject the subject.
+     * @param child the recipient.
+     */
+    private void notifyChildOfSubjectDeletion(final Subject subject, final ActorRef child) {
+        final var activeSubjectIdCount = activeSubjectIds.getOrDefault(subject.getId(), 0);
+        if (activeSubjectIdCount >= 2) {
+            // another actor took over the responsibility of child; terminate it.
+            log.debug("Terminating child <{}> of updated subject <{}>", child, subject);
+            child.tell(PoisonPill.getInstance(), ActorRef.noSender());
+        } else {
+            log.debug("Notifying child <{}> of deleted subject <{}>", child, subject);
+            child.tell(SubjectExpiryActor.Message.SUBJECT_DELETED, ActorRef.noSender());
+        }
+    }
+
     private Set<Subject> getSubjectsWithExpiryOrAnnouncements(final Policy policy) {
         if (policy.getLifecycle().filter(lifeCycle -> lifeCycle == PolicyLifecycle.ACTIVE).isPresent()) {
             return StreamSupport.stream(policy.spliterator(), false)
@@ -146,6 +171,26 @@ private Set<Subject> getSubjectsWithExpiryOrAnnouncements(final Policy policy) {
         }
     }
 
+    private void addActiveSubjectId(final Subject subject) {
+        activeSubjectIds.compute(subject.getId(), (k, count) -> {
+            if (count == null) {
+                return 1;
+            } else {
+                return count + 1;
+            }
+        });
+    }
+
+    private void removeActiveSubjectId(final Subject subject) {
+        activeSubjectIds.computeIfPresent(subject.getId(), (k, count) -> {
+            if (count <= 1) {
+                return null;
+            } else {
+                return count - 1;
+            }
+        });
+    }
+
     private static List<Subject> calculateDifference(final Set<Subject> minuend, final Set<Subject> subtrahend) {
         return minuend.stream()
                 .filter(subject -> !subtrahend.contains(subject))