merge: #1139

1139: fix: db transaction inconsistencies when deleting keys r=megglos a=megglos

## Description

The main cause of #1001 was the bug fixed via 77e162c . Within the same transaction you couldn't insert a previously deleted key, this happened as after processing the first processes complete command a new message start event correlation was triggered, failing as the previous active process instance was still in the database as the transaction wasn't commited and deleted keys of the transaction weren't checked.

I also noticed there was a bug in the `get` by key method, not checking the deletedKeys set either and fixed that too with bf9e1eb .

I verified the fix as well with the test case provided in the original issue, I decided against adding it though as the unit test covers the behavior of the db transaction that was the cause of the message start event test case failing.

## Related issues

<!-- Which issues are closed by this PR or are related -->

closes #1001

Co-authored-by: Meggle (Sebastian Bathke) <sebastian.bathke@camunda.com>

engine/src/main/java/io/camunda/zeebe/process/test/engine/db/InMemoryDbTransaction.java

@@ -73,6 +73,10 @@ public void put(final FullyQualifiedKey fullyQualifiedKey, final DbValue value)
   public byte[] get(final FullyQualifiedKey fullyQualifiedKey) {
     final Bytes key = fullyQualifiedKey.getKeyBytes();
 
+    if (deletedKeys.contains(key)) {
+      return null;
+    }
+
     final Bytes valueInCache = transactionCache.get(key);
 
     if (valueInCache != null) {
@@ -108,6 +112,7 @@ public InMemoryDbIterator newIterator() {
   @Override
   public boolean contains(final FullyQualifiedKey fullyQualifiedKey) {
     final Bytes keyBytes = fullyQualifiedKey.getKeyBytes();
-    return transactionCache.containsKey(keyBytes) || database.containsKey(keyBytes);
+    return !deletedKeys.contains(keyBytes)
+        && (transactionCache.containsKey(keyBytes) || database.containsKey(keyBytes));
   }
 }

engine/src/test/java/io/camunda/zeebe/process/test/engine/db/InMemoryZeebeDbTransactionTest.java

@@ -440,6 +440,49 @@ void shouldWriteAndDeleteSameKeyValuePairInTransaction() {
     assertThat(oneColumnFamily.exists(oneKey)).isFalse();
   }
 
+  @Test
+  void shouldAllowDeleteAndInsertInTransaction() throws Exception {
+    // given
+    oneKey.wrapLong(1);
+    oneValue.wrapLong(-1L);
+    twoValue.wrapLong(-2L);
+    oneColumnFamily.insert(oneKey, oneValue);
+    transactionContext.getCurrentTransaction().commit();
+
+    // when
+    transactionContext.runInTransaction(
+        () -> {
+          oneColumnFamily.deleteExisting(oneKey);
+          oneColumnFamily.insert(oneKey, twoValue);
+        });
+
+    // then
+    assertThat(oneColumnFamily.get(oneKey).getValue()).isEqualTo(twoValue.getValue());
+  }
+
+  @Test
+  void shouldNotGetByKeyIfDeletedInTransaction() throws Exception {
+    // given
+    oneKey.wrapLong(1);
+    oneValue.wrapLong(-1L);
+    twoValue.wrapLong(-2L);
+    oneColumnFamily.insert(oneKey, oneValue);
+    transactionContext.getCurrentTransaction().commit();
+
+    // when
+    transactionContext.runInTransaction(
+        () -> {
+          oneColumnFamily.deleteExisting(oneKey);
+
+          if (oneColumnFamily.get(oneKey) != null) {
+            fail("Should not be able to get deleted key.");
+          }
+        });
+
+    // then
+    assertThat(oneColumnFamily.get(oneKey)).isNull();
+  }
+
   @Test
   void shouldNotCommitOnError() {
     // given