Merge pull request eclipse-ditto#1863 from bosch-io/bugfix/thingCreat…

…e-policy-enrichment Fix enrichment of ThingCreated events with _policy
thfries · Jan 16, 2024 · c3c18ee · c3c18ee
2 parents c30d459 + 242aa1d
commit c3c18ee
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 2 deletions.
diff --git a/...rg/eclipse/ditto/internal/models/signalenrichment/DittoCachingSignalEnrichmentFacade.java b/...rg/eclipse/ditto/internal/models/signalenrichment/DittoCachingSignalEnrichmentFacade.java
@@ -219,8 +219,9 @@ private CompletableFuture<JsonObject> smartUpdateCachedObject(final SignalEnrich
                 final var nextExpectedThingEventsParameters =
                         new CachingParameters(fieldSelector, thingEvents, invalidateCacheOnPolicyChange,
                                 cachingParameters.minAcceptableSeqNr);
-                result = handleNextExpectedThingEvents(cacheKey, JsonObject.empty(), nextExpectedThingEventsParameters)
-                        .toCompletableFuture();
+                result = doCacheLookup(cacheKey, dittoHeaders).thenCompose(
+                        cachedJsonObject -> handleNextExpectedThingEvents(cacheKey, cachedJsonObject,
+                                nextExpectedThingEventsParameters));
             } else {
                 // there are twin events; perform smart update
                 result = doCacheLookup(cacheKey, dittoHeaders).thenCompose(

diff --git a/...rg/eclipse/ditto/thingsearch/service/persistence/write/streaming/EnforcementFlowTest.java b/...rg/eclipse/ditto/thingsearch/service/persistence/write/streaming/EnforcementFlowTest.java
@@ -253,6 +253,8 @@ public void computeThingCacheValueFromThingEvents() {
             sourceProbe.sendNext(inputMap);
             sourceProbe.sendComplete();
 
+            thingsProbe.expectMsgClass(Duration.apply(30, TimeUnit.SECONDS), SudoRetrieveThing.class);
+            thingsProbe.reply(SudoRetrieveThingResponse.of(JsonObject.empty(), DittoHeaders.empty()));
             // WHEN: policy is retrieved with up-to-date revisions
             policiesProbe.expectMsgClass(Duration.apply(30, TimeUnit.SECONDS), SudoRetrievePolicy.class);
             final var policy = Policy.newBuilder(policyId).setRevision(1).build();
@@ -319,6 +321,10 @@ public void computeThingCacheValueFromThingEventsWhenLastEventWasDeleted() {
             sourceProbe.sendNext(inputMap);
             sourceProbe.sendComplete();
 
+            // WHEN
+            thingsProbe.expectMsgClass(Duration.apply(30, TimeUnit.SECONDS), SudoRetrieveThing.class);
+            thingsProbe.reply(SudoRetrieveThingResponse.of(JsonObject.empty(), DittoHeaders.empty()));
+
             // THEN: the write model contains up-to-date revisions
             final AbstractWriteModel deleteModel = sinkProbe.expectNext().get(0);
             sinkProbe.expectComplete();