merge: #930

930: Don't iterate over deleted keys in a transaction r=remcowesterhoud a=remcowesterhoud

## Description

<!-- Please explain the changes you made here. -->
When iterating in an open transaction, we would create an iterator containing all entries in the database as well as the entries in the transaction cache. This ensures any keys added in the same transaction are also encountered during iterating.

We do not do anything with the keys that are deleted within the transaction. As a result, when iterating we will go over keys that are already deleted. This is wrong. The keys are deleted, so we should not encounter them anymore during iterating.

## Related issues

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

closes #876

<!-- Cut-off marker
_All lines under and including the cut-off marker will be removed from the merge commit message_

## Definition of Ready

Please check the items that apply, before requesting a review.

You can find more details about these items in our wiki page about [Pull Requests and Code Reviews](https://github.com/camunda/zeebe/wiki/Pull-Requests-and-Code-Reviews).

* [ ] I've reviewed my own code
* [ ] I've written a clear changelist description
* [ ] I've narrowly scoped my changes
* [ ] I've separated structural from behavioural changes
-->

## Definition of Done

<!-- Please check the items that apply, before merging or (if possible) before requesting a review. -->

_Not all items need to be done depending on the issue and the pull request._

Code changes:
* [x] The changes are backwards compatibility with previous versions
* [x] If it fixes a bug then PRs are created to backport the fix

Testing:
* [ ] There are unit/integration tests that verify all acceptance criterias of the issue
* [x] New tests are written to ensure backwards compatibility with further versions
* [ ] The behavior is tested manually

Documentation:
* [ ] Javadoc has been written
* [ ] The documentation is updated


Co-authored-by: Remco Westerhoud <[email protected]>

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

@@ -100,6 +100,7 @@ public InMemoryDbIterator newIterator() {
     final TreeMap<Bytes, Bytes> snapshot = new TreeMap<>();
     snapshot.putAll(database);
     snapshot.putAll(transactionCache);
+    deletedKeys.forEach(snapshot::remove);
 
     return new InMemoryDbIterator(snapshot);
   }

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

@@ -8,6 +8,7 @@
 package io.camunda.zeebe.process.test.engine.db;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
 
 import io.camunda.zeebe.db.ColumnFamily;
 import io.camunda.zeebe.db.TransactionContext;
@@ -492,6 +493,25 @@ void shouldWriteKeyAfterDeletion() {
     assertThat(oneColumnFamily.get(oneKey).getValue()).isEqualTo(-2);
   }
 
+  @Test
+  void shouldNotIterateOverDeletionsInTransaction() throws Exception {
+    // given
+    oneKey.wrapLong(1);
+    oneValue.wrapLong(-1L);
+    oneColumnFamily.insert(oneKey, oneValue);
+    transactionContext.getCurrentTransaction().commit();
+
+    // when - then
+    transactionContext.runInTransaction(
+        () -> {
+          oneColumnFamily.deleteExisting(oneKey);
+          oneColumnFamily.forEach(
+              (key, value) -> {
+                fail("Should not iterate over deleted keys");
+              });
+        });
+  }
+
   private enum ColumnFamilies {
     DEFAULT, // rocksDB needs a default column family
     ONE,