merge: #15199 #15249 #15262

15199: feat(client): provide the Java client's interfaces for migrating a process instance r=berkaycanbc a=berkaycanbc

## Description

* New command interface is created to define process instance migration steps
* A MigrationPlan object is added along with a builder implementation to allow reusing migration plans
* Noop implementation of process instance migration is implemented
* Tests added to verify building of the process instance migration request

## Related issues

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

closes #15161



15249: feat(topology): additional metrics for applying change operations r=oleschoenburg a=oleschoenburg

Adds counters for failed and applied operations as well as a gauge that times the time it takes to (successfully or not) apply an operation.

Relates to #15172 

15262: test(qa): await until new leaders during scale up or scale down r=deepthidevaki a=deepthidevaki

## Description

Found a flaky test in #15214 

The jobs were not activated because gateway does know the leader for a partition. This is because the existing test for complete topology passes while it still has the outdated topology with the old leader. 



Co-authored-by: berkaycanbc <berkay.can@camunda.com>
Co-authored-by: Ole Schönburg <ole.schoenburg@gmail.com>
Co-authored-by: Deepthi Devaki Akkoorath <deepthidevaki@gmail.com>

clients/java/src/main/java/io/camunda/zeebe/client/ZeebeClient.java

@@ -15,13 +15,15 @@
  */
 package io.camunda.zeebe.client;
 
+import io.camunda.zeebe.client.api.ExperimentalApi;
 import io.camunda.zeebe.client.api.command.BroadcastSignalCommandStep1;
 import io.camunda.zeebe.client.api.command.CancelProcessInstanceCommandStep1;
 import io.camunda.zeebe.client.api.command.CreateProcessInstanceCommandStep1;
 import io.camunda.zeebe.client.api.command.DeleteResourceCommandStep1;
 import io.camunda.zeebe.client.api.command.DeployProcessCommandStep1;
 import io.camunda.zeebe.client.api.command.DeployResourceCommandStep1;
 import io.camunda.zeebe.client.api.command.EvaluateDecisionCommandStep1;
+import io.camunda.zeebe.client.api.command.MigrateProcessInstanceCommandStep1;
 import io.camunda.zeebe.client.api.command.ModifyProcessInstanceCommandStep1;
 import io.camunda.zeebe.client.api.command.PublishMessageCommandStep1;
 import io.camunda.zeebe.client.api.command.ResolveIncidentCommandStep1;
@@ -167,6 +169,63 @@ static ZeebeClientCloudBuilderStep1 newCloudClientBuilder() {
    */
   ModifyProcessInstanceCommandStep1 newModifyProcessInstanceCommand(long processInstanceKey);
 
+  /**
+   * Command to migrate a process instance to a different process definition.
+   *
+   * <p>The migration command contains a migration plan. Migration plan contains
+   * targetProcessDefinitionKey to indicate which process definition to use for the migration.
+   * Mapping instructions for the migration describe how to map elements from the source process
+   * definition to the target process definition.
+   *
+   * <p>For example, let's consider we want to migrate process instance with key {@code 1}, target
+   * process definition key {@code 2}, a source process definition with a service task with id
+   * {@code "task1"} and the target process definition with a service task with id {@code "task2"}.
+   * The migration command could be:
+   *
+   * <pre>{@code
+   * {
+   *  "processInstanceKey": 1,
+   *  "migrationPlan": {
+   *   "targetProcessDefinitionKey": 2,
+   *   "mappingInstructions": [
+   *    {
+   *     "sourceElementId": "task1",
+   *     "targetElementId": "task2"
+   *    }
+   *   ]
+   *  }
+   * }
+   * }</pre>
+   *
+   * <pre>
+   *
+   * zeebeClient
+   *  .newMigrateProcessInstanceCommand(1L)
+   *  .migrationPlan(2L)
+   *  .addMappingInstruction("element1", "element2")
+   *  .addMappingInstruction("element3", "element4")
+   *  .send();
+   * </pre>
+   *
+   * <pre>
+   * final MigrationPlan migrationPlan =
+   *         MigrationPlan.newBuilder()
+   *             .withTargetProcessDefinitionKey(2L)
+   *             .addMappingInstruction("element1", "element2")
+   *             .addMappingInstruction("element3", "element4")
+   *             .build();
+   * zeebeClient
+   *  .newMigrateProcessInstanceCommand(1L)
+   *  .migrationPlan(migrationPlan)
+   *  .send();
+   * </pre>
+   *
+   * @param processInstanceKey the key which refers to the process instance to migrate
+   * @return a builder for the command
+   */
+  @ExperimentalApi("https://github.com/camunda/zeebe/issues/14907")
+  MigrateProcessInstanceCommandStep1 newMigrateProcessInstanceCommand(long processInstanceKey);
+
   /**
    * Command to cancel a process instance.
    *

clients/java/src/main/java/io/camunda/zeebe/client/api/command/MigrateProcessInstanceCommandStep1.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright © 2017 camunda services GmbH (info@camunda.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.camunda.zeebe.client.api.command;
+
+import io.camunda.zeebe.client.api.ExperimentalApi;
+import io.camunda.zeebe.client.api.response.MigrateProcessInstanceResponse;
+
+@ExperimentalApi("https://github.com/camunda/zeebe/issues/14907")
+public interface MigrateProcessInstanceCommandStep1 {
+
+  /**
+   * Create a MigrationPlan (TODO - link to actual gRPC object) for the given target process
+   * definition key.
+   *
+   * @param targetProcessDefinitionKey the key of the target process definition
+   * @return the builder for this command
+   */
+  MigrateProcessInstanceCommandStep2 migrationPlan(final long targetProcessDefinitionKey);
+
+  /**
+   * Use the provided MigrationPlan from the given {@link MigrationPlan} object.
+   *
+   * <p>Example MigrationPlan object creation:
+   *
+   * <pre>
+   * final MigrationPlan migrationPlan =
+   *         MigrationPlan.newBuilder()
+   *             .withTargetProcessDefinitionKey(2L)
+   *             .addMappingInstruction("element1", "element2")
+   *             .addMappingInstruction("element3", "element4")
+   *             .build();
+   * </pre>
+   *
+   * @param migrationPlan the object that contains migration plan data
+   * @return the builder for this command
+   */
+  MigrateProcessInstanceCommandFinalStep migrationPlan(final MigrationPlan migrationPlan);
+
+  interface MigrateProcessInstanceCommandStep2 {
+    /**
+     * Add a MappingInstruction for the element that will be migrated and its target element id in
+     * the target process definition.
+     *
+     * @param sourceElementId the element id to migrate from
+     * @param targetElementId the element id to migrate into
+     * @return the builder for this command
+     */
+    MigrateProcessInstanceCommandFinalStep addMappingInstruction(
+        final String sourceElementId, final String targetElementId);
+  }
+
+  interface MigrateProcessInstanceCommandFinalStep
+      extends MigrateProcessInstanceCommandStep2, FinalCommandStep<MigrateProcessInstanceResponse> {
+
+    /**
+     * Add a MappingInstruction for the element that will be migrated and its target element id in
+     * the target process definition. This method allows to add more than one mapping instructions
+     * to the migration plan.
+     *
+     * @param sourceElementId the element id to migrate from
+     * @param targetElementId the element id to migrate into
+     * @return the builder for this command
+     */
+    @Override
+    MigrateProcessInstanceCommandFinalStep addMappingInstruction(
+        final String sourceElementId, final String targetElementId);
+  }
+}

clients/java/src/main/java/io/camunda/zeebe/client/api/command/MigrationPlan.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright © 2017 camunda services GmbH (info@camunda.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.camunda.zeebe.client.api.command;
+
+import io.camunda.zeebe.client.api.command.MigrationPlanBuilderImpl.MappingInstruction;
+import java.util.List;
+
+public interface MigrationPlan {
+
+  /** Create a new migration plan builder to build {@link MigrationPlan} object */
+  static MigrationPlanBuilderStep1 newBuilder() {
+    return new MigrationPlanBuilderImpl();
+  }
+
+  /**
+   * Get the key of target process definition. targetProcessDefinitionKey indicates which process
+   * definition to use for the migration.
+   *
+   * @return the target process definition key
+   */
+  public long getTargetProcessDefinitionKey();
+
+  /**
+   * Get mapping instructions to the migration for describing how to map elements from the source
+   * process definition to the target process definition.
+   *
+   * @return list of mapping instructions
+   */
+  public List<MappingInstruction> getMappingInstructions();
+
+  interface MigrationPlanBuilderStep1 {
+
+    /**
+     * Set the key of target process definition. targetProcessDefinitionKey indicates which process
+     * definition to use for the migration.
+     *
+     * @return the next step of the builder
+     */
+    MigrationPlanBuilderStep2 withTargetProcessDefinitionKey(final long targetProcessDefinitionKey);
+  }
+
+  interface MigrationPlanBuilderStep2 {
+
+    /**
+     * Add a mapping instruction to the migration for describing how to map elements from the source
+     * process definition to the target process definition.
+     *
+     * <p>For example, let's consider a source process definition with a service task with id {@code
+     * "task1"} and the target process definition with a service task with id {@code "task2"}. The
+     * mapping instruction could be:
+     *
+     * <pre>{@code
+     * {
+     *   "sourceElementId": "task1",
+     *   "targetElementId": "task2"
+     * }
+     * }</pre>
+     *
+     * This mapping would migrate instances of the service task with id {@code "task1"} to the
+     * service task with id {@code "task2"}.
+     *
+     * @param sourceElementId element to migrate
+     * @param targetElementId element to migrate into
+     * @return the next step of the builder
+     */
+    MigrationPlanBuilderFinalStep addMappingInstruction(
+        final String sourceElementId, final String targetElementId);
+  }
+
+  interface MigrationPlanBuilderFinalStep extends MigrationPlanBuilderStep2 {
+
+    /**
+     * Build the {@link MigrationPlan} object after filling the object with migration data
+     *
+     * @return a reusable migration plan
+     */
+    MigrationPlan build();
+  }
+}

clients/java/src/main/java/io/camunda/zeebe/client/api/command/MigrationPlanBuilderImpl.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright © 2017 camunda services GmbH (info@camunda.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.camunda.zeebe.client.api.command;
+
+import io.camunda.zeebe.client.api.command.MigrationPlan.MigrationPlanBuilderFinalStep;
+import io.camunda.zeebe.client.api.command.MigrationPlan.MigrationPlanBuilderStep1;
+import io.camunda.zeebe.client.api.command.MigrationPlan.MigrationPlanBuilderStep2;
+import java.util.ArrayList;
+import java.util.List;
+
+public final class MigrationPlanBuilderImpl
+    implements MigrationPlanBuilderStep1, MigrationPlanBuilderStep2, MigrationPlanBuilderFinalStep {
+
+  private long targetProcessDefinitionKey;
+  private final List<MappingInstruction> mappingInstructions;
+
+  public MigrationPlanBuilderImpl() {
+    mappingInstructions = new ArrayList<>();
+  }
+
+  @Override
+  public MigrationPlanBuilderStep2 withTargetProcessDefinitionKey(
+      final long targetProcessDefinitionKey) {
+    this.targetProcessDefinitionKey = targetProcessDefinitionKey;
+    return this;
+  }
+
+  @Override
+  public MigrationPlanBuilderFinalStep addMappingInstruction(
+      final String sourceElementId, final String targetElementId) {
+    mappingInstructions.add(new MappingInstruction(sourceElementId, targetElementId));
+    return this;
+  }
+
+  @Override
+  public MigrationPlan build() {
+    return new MigrationPlanImpl(targetProcessDefinitionKey, mappingInstructions);
+  }
+
+  public static class MappingInstruction {
+    private final String sourceElementId;
+    private final String targetElementId;
+
+    public MappingInstruction(final String sourceElementId, final String targetElementId) {
+      this.sourceElementId = sourceElementId;
+      this.targetElementId = targetElementId;
+    }
+
+    public String getSourceElementId() {
+      return sourceElementId;
+    }
+
+    public String getTargetElementId() {
+      return targetElementId;
+    }
+  }
+}

clients/java/src/main/java/io/camunda/zeebe/client/api/command/MigrationPlanImpl.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright © 2017 camunda services GmbH (info@camunda.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.camunda.zeebe.client.api.command;
+
+import io.camunda.zeebe.client.api.command.MigrationPlanBuilderImpl.MappingInstruction;
+import java.util.List;
+
+public final class MigrationPlanImpl implements MigrationPlan {
+
+  final long targetProcessDefinitionKey;
+  final List<MappingInstruction> mappingInstructions;
+
+  public MigrationPlanImpl(
+      final long targetProcessDefinitionKey, final List<MappingInstruction> mappingInstructions) {
+    this.targetProcessDefinitionKey = targetProcessDefinitionKey;
+    this.mappingInstructions = mappingInstructions;
+  }
+
+  @Override
+  public long getTargetProcessDefinitionKey() {
+    return targetProcessDefinitionKey;
+  }
+
+  @Override
+  public List<MappingInstruction> getMappingInstructions() {
+    return mappingInstructions;
+  }
+}

clients/java/src/main/java/io/camunda/zeebe/client/api/response/MigrateProcessInstanceResponse.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright © 2017 camunda services GmbH (info@camunda.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.camunda.zeebe.client.api.response;
+
+import io.camunda.zeebe.client.api.ExperimentalApi;
+
+@ExperimentalApi("https://github.com/camunda/zeebe/issues/14907")
+public interface MigrateProcessInstanceResponse {}