Bump Zeebe to version 8.0.0-rc1

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

@@ -7,7 +7,12 @@
  */
 package io.camunda.zeebe.process.test.engine.db;
 
-import io.camunda.zeebe.db.*;
+import io.camunda.zeebe.db.ColumnFamily;
+import io.camunda.zeebe.db.DbKey;
+import io.camunda.zeebe.db.DbValue;
+import io.camunda.zeebe.db.KeyValuePairVisitor;
+import io.camunda.zeebe.db.TransactionContext;
+import io.camunda.zeebe.db.ZeebeDbInconsistentException;
 import io.camunda.zeebe.util.buffer.BufferUtil;
 import java.util.Iterator;
 import java.util.Map;
@@ -50,7 +55,37 @@ private void ensureInOpenTransaction(
   }
 
   @Override
-  public void put(final KeyType key, final ValueType value) {
+  public void insert(final KeyType key, final ValueType value) {
+    ensureInOpenTransaction(
+        context,
+        state -> {
+          final FullyQualifiedKey fullyQualifiedKey = new FullyQualifiedKey(columnFamily, key);
+          if (state.contains(fullyQualifiedKey)) {
+            throw new ZeebeDbInconsistentException(
+                "Key " + keyInstance + " in ColumnFamily " + columnFamily + " already exists");
+          } else {
+            state.put(fullyQualifiedKey, value);
+          }
+        });
+  }
+
+  @Override
+  public void update(final KeyType key, final ValueType value) {
+    ensureInOpenTransaction(
+        context,
+        state -> {
+          final FullyQualifiedKey fullyQualifiedKey = new FullyQualifiedKey(columnFamily, key);
+          if (state.contains(fullyQualifiedKey)) {
+            state.put(fullyQualifiedKey, value);
+          } else {
+            throw new ZeebeDbInconsistentException(
+                "Key " + keyInstance + " in ColumnFamily " + columnFamily + " does not exist");
+          }
+        });
+  }
+
+  @Override
+  public void upsert(final KeyType key, final ValueType value) {
     ensureInOpenTransaction(
         context, state -> state.put(new FullyQualifiedKey(columnFamily, key), value));
   }
@@ -71,17 +106,6 @@ public ValueType get(final KeyType key) {
     return null;
   }
 
-  private DirectBuffer getValue(final InMemoryDbState state, final DbKey key) {
-    final FullyQualifiedKey fullyQualifiedKey = new FullyQualifiedKey(columnFamily, key);
-    final byte[] value = state.get(fullyQualifiedKey);
-
-    if (value != null) {
-      return BufferUtil.wrapArray(value);
-    } else {
-      return null;
-    }
-  }
-
   @Override
   public void forEach(final Consumer<ValueType> consumer) {
     forEach(context, (key, value) -> consumer.accept(value));
@@ -92,21 +116,11 @@ public void forEach(final BiConsumer<KeyType, ValueType> consumer) {
     forEach(context, consumer);
   }
 
-  private void forEach(
-      final TransactionContext context, final BiConsumer<KeyType, ValueType> consumer) {
-    whileEqualPrefix(context, keyInstance, valueInstance, consumer);
-  }
-
   @Override
   public void whileTrue(final KeyValuePairVisitor<KeyType, ValueType> visitor) {
     whileTrue(context, visitor);
   }
 
-  private void whileTrue(
-      final TransactionContext context, final KeyValuePairVisitor<KeyType, ValueType> visitor) {
-    whileEqualPrefix(context, DbNullKey.INSTANCE, keyInstance, valueInstance, visitor);
-  }
-
   @Override
   public void whileEqualPrefix(
       final DbKey keyPrefix, final BiConsumer<KeyType, ValueType> visitor) {
@@ -119,6 +133,74 @@ public void whileEqualPrefix(
     whileEqualPrefix(context, keyPrefix, keyInstance, valueInstance, visitor);
   }
 
+  @Override
+  public void deleteExisting(final KeyType key) {
+    ensureInOpenTransaction(
+        context,
+        state -> {
+          final FullyQualifiedKey fullyQualifiedKey = new FullyQualifiedKey(columnFamily, key);
+          if (state.contains(fullyQualifiedKey)) {
+            state.delete(fullyQualifiedKey);
+          } else {
+            throw new ZeebeDbInconsistentException(
+                "Key " + keyInstance + " in ColumnFamily " + columnFamily + " does not exist");
+          }
+        });
+  }
+
+  @Override
+  public void deleteIfExists(final KeyType key) {
+    ensureInOpenTransaction(
+        context,
+        state -> {
+          final FullyQualifiedKey fullyQualifiedKey = new FullyQualifiedKey(columnFamily, key);
+          state.delete(fullyQualifiedKey);
+        });
+  }
+
+  @Override
+  public boolean exists(final KeyType key) {
+    final AtomicBoolean exists = new AtomicBoolean(true);
+
+    ensureInOpenTransaction(context, state -> exists.set(getValue(state, key) != null));
+
+    return exists.get();
+  }
+
+  @Override
+  public boolean isEmpty() {
+    final AtomicBoolean isEmpty = new AtomicBoolean(true);
+    whileEqualPrefix(
+        DbNullKey.INSTANCE,
+        (key, value) -> {
+          isEmpty.set(false);
+          return false;
+        });
+
+    return isEmpty.get();
+  }
+
+  private DirectBuffer getValue(final InMemoryDbState state, final DbKey key) {
+    final FullyQualifiedKey fullyQualifiedKey = new FullyQualifiedKey(columnFamily, key);
+    final byte[] value = state.get(fullyQualifiedKey);
+
+    if (value != null) {
+      return BufferUtil.wrapArray(value);
+    } else {
+      return null;
+    }
+  }
+
+  private void forEach(
+      final TransactionContext context, final BiConsumer<KeyType, ValueType> consumer) {
+    whileEqualPrefix(context, keyInstance, valueInstance, consumer);
+  }
+
+  private void whileTrue(
+      final TransactionContext context, final KeyValuePairVisitor<KeyType, ValueType> visitor) {
+    whileEqualPrefix(context, DbNullKey.INSTANCE, keyInstance, valueInstance, visitor);
+  }
+
   private void whileEqualPrefix(
       final TransactionContext context,
       final KeyType keyInstance,
@@ -171,34 +253,6 @@ private void whileEqualPrefix(
                 }));
   }
 
-  @Override
-  public void delete(final KeyType key) {
-    final FullyQualifiedKey fullyQualifiedKey = new FullyQualifiedKey(columnFamily, key);
-    ensureInOpenTransaction(context, state -> state.delete(fullyQualifiedKey));
-  }
-
-  @Override
-  public boolean exists(final KeyType key) {
-    final AtomicBoolean exists = new AtomicBoolean(true);
-
-    ensureInOpenTransaction(context, state -> exists.set(getValue(state, key) != null));
-
-    return exists.get();
-  }
-
-  @Override
-  public boolean isEmpty() {
-    final AtomicBoolean isEmpty = new AtomicBoolean(true);
-    whileEqualPrefix(
-        DbNullKey.INSTANCE,
-        (key, value) -> {
-          isEmpty.set(false);
-          return false;
-        });
-
-    return isEmpty.get();
-  }
-
   private static final class Visitor<KeyType extends DbKey, ValueType extends DbValue>
       implements KeyValuePairVisitor<KeyType, ValueType> {
 

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

@@ -17,4 +17,6 @@ interface InMemoryDbState {
   void delete(FullyQualifiedKey fullyQualifiedKey);
 
   InMemoryDbIterator newIterator();
+
+  boolean contains(FullyQualifiedKey fullyQualifiedKey);
 }

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

@@ -103,4 +103,10 @@ public InMemoryDbIterator newIterator() {
 
     return new InMemoryDbIterator(snapshot);
   }
+
+  @Override
+  public boolean contains(final FullyQualifiedKey fullyQualifiedKey) {
+    final Bytes keyBytes = fullyQualifiedKey.getKeyBytes();
+    return transactionCache.containsKey(keyBytes) || database.containsKey(keyBytes);
+  }
 }

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

@@ -9,6 +9,7 @@
 
 import io.camunda.zeebe.db.TransactionContext;
 import io.camunda.zeebe.db.TransactionOperation;
+import io.camunda.zeebe.db.ZeebeDbInconsistentException;
 import io.camunda.zeebe.db.ZeebeDbTransaction;
 import java.util.TreeMap;
 
@@ -27,6 +28,8 @@ public void runInTransaction(final TransactionOperation operations) {
       } else {
         runInNewTransaction(operations);
       }
+    } catch (final ZeebeDbInconsistentException e) {
+      throw e;
     } catch (final Exception e) {
       throw new RuntimeException(
           "Unexpected error occurred during zeebe db transaction operation.", e);

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

@@ -45,7 +45,7 @@ void shouldPutValue() {
     value.wrapString("baring");
 
     // when
-    columnFamily.put(key, value);
+    columnFamily.upsert(key, value);
     value.wrapString("yes");
 
     // then
@@ -223,6 +223,6 @@ void shouldThrowExceptionOnMultipleNestedWhileEqualPrefix() {
   private void putKeyValuePair(final String key, final String value) {
     this.key.wrapString(key);
     this.value.wrapString(value);
-    columnFamily.put(this.key, this.value);
+    columnFamily.upsert(this.key, this.value);
   }
 }