[eclipse-ditto#890] Fix off-by-1 error in revision of scheduled Subje…

…ctDeleted events; optimize PolicyEventForwarder to not query database in the absence of policy changes. Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
bosch-io · Dec 14, 2020 · a685b26 · a685b26
1 parent cb12f4d
commit a685b26
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 4 deletions.
diff --git a/...n/java/org/eclipse/ditto/services/policies/persistence/actors/PolicyPersistenceActor.java b/...n/java/org/eclipse/ditto/services/policies/persistence/actors/PolicyPersistenceActor.java
@@ -258,7 +258,7 @@ private Optional<SubjectDeleted> calculateSubjectDeletedEventOfOldestExpiredSubj
                         SubjectDeleted.of(policyId,
                                 pair.first(),
                                 pair.second().getId(),
-                                getRevisionNumber(),
+                                getRevisionNumber() + 1L,
                                 eventTimestamp,
                                 eventDittoHeaders
                         )

diff --git a/...va/org/eclipse/ditto/services/policies/persistence/actors/PolicyPersistenceActorTest.java b/...va/org/eclipse/ditto/services/policies/persistence/actors/PolicyPersistenceActorTest.java
@@ -817,13 +817,13 @@ public void createPolicyWith2SubjectsWithExpiry() {
             // GIVEN: a Policy is created with 2 subjects having an "expiry" date
             final CreatePolicy createPolicyCommand = CreatePolicy.of(policy, dittoHeadersV2);
             underTest.tell(createPolicyCommand, getRef());
-            final CreatePolicyResponse createPolicy1Response = expectMsgClass(CreatePolicyResponse.class);
+            expectMsgClass(CreatePolicyResponse.class);
 
             // THEN: a PolicyCreated event should be emitted
             final DistributedPubSubMediator.Publish policyCreatedPublish =
                     pubSubMediatorTestProbe.expectMsgClass(DistributedPubSubMediator.Publish.class);
             assertThat(policyCreatedPublish.msg()).isInstanceOf(PolicyCreated.class);
-
+            assertThat(((PolicyCreated) policyCreatedPublish.msg()).getRevision()).isEqualTo(1L);
 
             // THEN: subject1 is deleted after expiry
             final long secondsToAdd = 10 - (expiryInstant.getEpochSecond() % 10);
@@ -838,6 +838,7 @@ public void createPolicyWith2SubjectsWithExpiry() {
                             DistributedPubSubMediator.Publish.class);
             assertThat(subject1Deleted.msg()).isInstanceOf(SubjectDeleted.class);
             assertThat(((SubjectDeleted) subject1Deleted.msg()).getSubjectId()).isEqualTo(subject1.getId());
+            assertThat(((SubjectDeleted) subject1Deleted.msg()).getRevision()).isEqualTo(2L);
             assertThat(pubSubMediatorTestProbe.expectMsgClass(DistributedPubSubMediator.Publish.class).msg())
                     .isInstanceOf(PolicyTag.class);
 
@@ -846,6 +847,7 @@ public void createPolicyWith2SubjectsWithExpiry() {
                     pubSubMediatorTestProbe.expectMsgClass(DistributedPubSubMediator.Publish.class);
             assertThat(subject2Deleted.msg()).isInstanceOf(SubjectDeleted.class);
             assertThat(((SubjectDeleted) subject2Deleted.msg()).getSubjectId()).isEqualTo(subject2.getId());
+            assertThat(((SubjectDeleted) subject2Deleted.msg()).getRevision()).isEqualTo(3L);
 
             // THEN: the policy has no subjects. (TODO: rationalize or change this behavior.)
             underTest.tell(RetrievePolicy.of(policy.getEntityId().orElseThrow(), DittoHeaders.empty()), getRef());

diff --git a/...main/java/org/eclipse/ditto/services/thingsearch/updater/actors/PolicyEventForwarder.java b/...main/java/org/eclipse/ditto/services/thingsearch/updater/actors/PolicyEventForwarder.java
@@ -198,7 +198,12 @@ private void terminateStream() {
     @SuppressWarnings("unchecked")
     private Source<PolicyReferenceTag, NotUsed> mapDumpResult(final Object dumpResult) {
         if (dumpResult instanceof Map) {
-            return persistence.getPolicyReferenceTags((Map<PolicyId, Long>) dumpResult);
+            final Map<PolicyId, Long> map = (Map<PolicyId, Long>) dumpResult;
+            if (map.isEmpty()) {
+                return Source.empty();
+            } else {
+                return persistence.getPolicyReferenceTags(map);
+            }
         } else {
             if (dumpResult instanceof Throwable) {
                 log.error((Throwable) dumpResult, "dump failed");