fix(engine): allow retried dmn resource distribution

This fixes critical a critical bug, where a decision and/or drg can be
inserted twice due to retry logic for inter-partition distribution of
deployments. To counter act, this allows an overwrite of existing
values.

See #9877 and DeploymentClusteredTest.shouldDistributeDmnResourceOnRetry.

(cherry picked from commit 0e60752)

engine/src/main/java/io/camunda/zeebe/engine/state/deployment/DbDecisionState.java

@@ -146,11 +146,11 @@ private Optional<PersistedDecision> findDecisionByKey(final long decisionKey) {
   public void storeDecisionRecord(final DecisionRecord record) {
     dbDecisionKey.wrapLong(record.getDecisionKey());
     dbPersistedDecision.wrap(record);
-    decisionsByKey.insert(dbDecisionKey, dbPersistedDecision);
+    decisionsByKey.upsert(dbDecisionKey, dbPersistedDecision);
 
     dbDecisionKey.wrapLong(record.getDecisionKey());
     dbDecisionRequirementsKey.wrapLong(record.getDecisionRequirementsKey());
-    decisionKeyByDecisionRequirementsKey.insert(
+    decisionKeyByDecisionRequirementsKey.upsert(
         dbDecisionRequirementsKeyAndDecisionKey, DbNil.INSTANCE);
 
     updateLatestDecisionVersion(record);
@@ -160,7 +160,7 @@ public void storeDecisionRecord(final DecisionRecord record) {
   public void storeDecisionRequirements(final DecisionRequirementsRecord record) {
     dbDecisionRequirementsKey.wrapLong(record.getDecisionRequirementsKey());
     dbPersistedDecisionRequirements.wrap(record);
-    decisionRequirementsByKey.insert(dbDecisionRequirementsKey, dbPersistedDecisionRequirements);
+    decisionRequirementsByKey.upsert(dbDecisionRequirementsKey, dbPersistedDecisionRequirements);
 
     updateLatestDecisionRequirementsVersion(record);
   }
@@ -185,7 +185,7 @@ private void updateDecisionAsLatestVersion(final DecisionRecord record) {
   private void insertDecisionAsLatestVersion(final DecisionRecord record) {
     dbDecisionId.wrapBuffer(record.getDecisionIdBuffer());
     dbDecisionKey.wrapLong(record.getDecisionKey());
-    latestDecisionKeysByDecisionId.insert(dbDecisionId, fkDecision);
+    latestDecisionKeysByDecisionId.upsert(dbDecisionId, fkDecision);
   }
 
   private void updateLatestDecisionRequirementsVersion(final DecisionRequirementsRecord record) {
@@ -209,6 +209,6 @@ private void updateDecisionRequirementsAsLatestVersion(final DecisionRequirement
   private void insertDecisionRequirementsAsLatestVersion(final DecisionRequirementsRecord record) {
     dbDecisionRequirementsId.wrapBuffer(record.getDecisionRequirementsIdBuffer());
     dbDecisionRequirementsKey.wrapLong(record.getDecisionRequirementsKey());
-    latestDecisionRequirementsKeysById.insert(dbDecisionRequirementsId, fkDecisionRequirements);
+    latestDecisionRequirementsKeysById.upsert(dbDecisionRequirementsId, fkDecisionRequirements);
   }
 }