fix: double delete events are handled correctly

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [unreleased]
 ### Fixed
+- `DefaultEntityViewUpdater` to handle receiving double delete events correctly.
+
+## [2.0.0] 2023-11-02
+### Fixed
 - KafkaEventReceiver progress logger will actually log the process during normal application bootstrapping.
 - Fixing test dependencies to be able to run on modern Mac hardware, Java 17 and Docker Desktop versions
 ### Updated [BREAKING CHANGES]

state/core/src/main/java/com/expediagroup/streamplatform/streamregistry/state/DefaultEntityViewUpdater.java

@@ -91,7 +91,7 @@ private <K extends Entity.Key<S>, S extends Specification> Entity<K, S> update(S
   }
 
   private <K extends Entity.Key<S>, S extends Specification> Entity<K, S> delete(SpecificationDeletionEvent<K, S> event) {
-    val oldEntity = (Entity<K, S>) getExistingEntity(event.getKey());
+    val oldEntity = (Entity<K, S>) getEntity(event.getKey());
     entities.put(event.getKey(), deleted(oldEntity));
     log.debug("Deleted entity for {}", event.getKey());
     return oldEntity;
@@ -115,4 +115,22 @@ private <K extends Entity.Key<S>, S extends Specification> Entity<K, S> delete(S
       .map(it -> it.entity)
       .orElse(null);
   }
+
+  /**
+   * There is a chance the entity will have already been deleted. Only use this method if you don't care.
+   */
+  private Entity<?, ?> getEntity(Entity.Key<?> key) {
+    val stateValue = Optional.ofNullable(entities.get(key));
+    stateValue.ifPresent(it -> {
+        if (it.deleted) {
+          log.debug("Found deleted entity for key={}", key);
+        } else {
+          log.debug("Found entity for key={}", key);
+        }
+      }
+    );
+    return stateValue
+      .map(it -> it.entity)
+      .orElse(null);
+  }
 }

state/core/src/test/java/com/expediagroup/streamplatform/streamregistry/state/EntityViewUpdaterTest.java

@@ -140,6 +140,25 @@ public void deleteSpecificationEvent() {
     assertThat(entities, hasEntry(key, deleted(entity.withStatus(oldStatus))));
   }
 
+  @Test
+  public void doubleDeleteSpecificationEvent() {
+    val previousEntity = underTest.update(specificationEvent);
+
+    assertThat(previousEntity, is(nullValue()));
+    assertThat(entities, is(aMapWithSize(1)));
+    assertThat(entities, hasEntry(key, existing(entity.withStatus(oldStatus))));
+
+    val deletedEntity = underTest.update(specificationDeletionEvent);
+    assertThat(deletedEntity, is(entity.withStatus(oldStatus)));
+    assertThat(entities, is(aMapWithSize(1)));
+    assertThat(entities, hasEntry(key, deleted(entity.withStatus(oldStatus))));
+
+    val doubleDeletedEntity = underTest.update(specificationDeletionEvent);
+    assertThat(doubleDeletedEntity, is(entity.withStatus(oldStatus)));
+    assertThat(entities, is(aMapWithSize(1)));
+    assertThat(entities, hasEntry(key, deleted(entity.withStatus(oldStatus))));
+  }
+
   @Test
   public void deleteStatusEvent() {
     entities.put(key, existing(oldEntity));

state/it/src/test/java/com/expediagroup/streamplatform/streamregistry/state/it/AgentIT.java

@@ -184,6 +184,21 @@ public void testDeletedEntities() {
       assertThat(dummyAgent.events, hasItem(Pair.of(data.getEntity(), specificationDeletion(data.getKey()))));  // the agent would be expected to handle the deletion event.
     });
 
+    // this shouldn't happen very often, but sending the same delete event should result in the old deleted entity still being available.
+    sendSync(kafkaEventSender, specificationDeletion(data.getKey()));
+    await.untilAsserted(() -> {
+      // entity no longer exists
+      assertThat(domainEvents(entityView), hasSize(0));
+      // entity is marked as deleted
+      assertThat(deletedDomainEvents(entityView), is(aMapWithSize(1)));
+      assertThat(deletedDomainEvents(entityView), hasEntry(data.getKey(), Optional.of(data.getEntity())));
+
+      // onEvent has been called for the deleted entity twice
+      assertThat(dummyAgent.events, hasSize(3));
+      assertThat(dummyAgent.events, hasItem(Pair.of(null, data.getSpecificationEvent())));
+      assertThat(dummyAgent.events, hasItem(Pair.of(data.getEntity(), specificationDeletion(data.getKey()))));  // the agent would be expected to handle the deletion event.
+    });
+
     // purge would be called by the agent after the delete has been handled
     entityView.purgeDeleted(data.getKey());
     // the delete is removed from everywhere