merge: #10711

10711: [Backport stable/8.1] Distribute deployment in post commit tasks r=remcowesterhoud a=backport-action # Description Backport of #10689 to `stable/8.1`. relates to #9964 Co-authored-by: Remco Westerhoud <remco@westerhoud.nl>
camunda · Oct 13, 2022 · 533e22d · 533e22d
2 parents f9e2d3d + 918f864
commit 533e22d
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 3 deletions.
diff --git a/...rc/main/java/io/camunda/zeebe/engine/processing/deployment/DeploymentCreateProcessor.java b/...rc/main/java/io/camunda/zeebe/engine/processing/deployment/DeploymentCreateProcessor.java
@@ -112,7 +112,7 @@ public void processRecord(
 
       stateWriter.appendFollowUpEvent(key, DeploymentIntent.CREATED, deploymentEvent);
 
-      deploymentDistributionBehavior.distributeDeployment(deploymentEvent, key);
+      deploymentDistributionBehavior.distributeDeployment(deploymentEvent, key, sideEffects);
       messageStartEventSubscriptionManager.tryReOpenMessageStartEventSubscription(
           deploymentEvent, stateWriter);
 

diff --git a/...camunda/zeebe/engine/processing/deployment/distribute/DeploymentDistributionBehavior.java b/...camunda/zeebe/engine/processing/deployment/distribute/DeploymentDistributionBehavior.java
@@ -7,6 +7,7 @@
  */
 package io.camunda.zeebe.engine.processing.deployment.distribute;
 
+import io.camunda.zeebe.engine.processing.streamprocessor.sideeffect.SideEffectQueue;
 import io.camunda.zeebe.engine.processing.streamprocessor.writers.StateWriter;
 import io.camunda.zeebe.engine.processing.streamprocessor.writers.Writers;
 import io.camunda.zeebe.protocol.Protocol;
@@ -45,7 +46,10 @@ public DeploymentDistributionBehavior(
     stateWriter = writers.state();
   }
 
-  public void distributeDeployment(final DeploymentRecord deploymentEvent, final long key) {
+  public void distributeDeployment(
+      final DeploymentRecord deploymentEvent,
+      final long key,
+      final SideEffectQueue sideEffectQueue) {
     final var copiedDeploymentBuffer = BufferUtil.createCopy(deploymentEvent);
 
     otherPartitions.forEach(
@@ -54,7 +58,11 @@ public void distributeDeployment(final DeploymentRecord deploymentEvent, final l
           stateWriter.appendFollowUpEvent(
               key, DeploymentDistributionIntent.DISTRIBUTING, deploymentDistributionRecord);
 
-          distributeDeploymentToPartition(key, partitionId, copiedDeploymentBuffer);
+          sideEffectQueue.add(
+              () -> {
+                distributeDeploymentToPartition(key, partitionId, copiedDeploymentBuffer);
+                return true;
+              });
         });
 
     if (otherPartitions.isEmpty()) {