fix: extract msg start subscription migration task

engine/src/main/java/io/camunda/zeebe/engine/state/migration/DbMigratorImpl.java

@@ -11,6 +11,7 @@
 import io.camunda.zeebe.engine.state.migration.to_8_2.DecisionMigration;
 import io.camunda.zeebe.engine.state.migration.to_8_2.DecisionRequirementsMigration;
 import io.camunda.zeebe.engine.state.migration.to_8_3.MultiTenancyDecisionStateMigration;
+import io.camunda.zeebe.engine.state.migration.to_8_3.MultiTenancyMessageStartEventSubscriptionStateMigration;
 import io.camunda.zeebe.engine.state.migration.to_8_3.MultiTenancyMessageStateMigration;
 import io.camunda.zeebe.engine.state.migration.to_8_3.MultiTenancyMigration;
 import io.camunda.zeebe.engine.state.migration.to_8_3.MultiTenancyProcessStateMigration;
@@ -42,6 +43,7 @@ public class DbMigratorImpl implements DbMigrator {
           new MultiTenancyProcessStateMigration(),
           new MultiTenancyDecisionStateMigration(),
           new MultiTenancyMessageStateMigration(),
+          new MultiTenancyMessageStartEventSubscriptionStateMigration(),
           new MultiTenancyMigration());
   // Be mindful of https://github.com/camunda/zeebe/issues/7248. In particular, that issue
   // should be solved first, before adding any migration that can take a long time

engine/src/main/java/io/camunda/zeebe/engine/state/migration/to_8_3/MultiTenancyMessageStartEventSubscriptionStateMigration.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
+ * one or more contributor license agreements. See the NOTICE file distributed
+ * with this work for additional information regarding copyright ownership.
+ * Licensed under the Zeebe Community License 1.1. You may not use this file
+ * except in compliance with the Zeebe Community License 1.1.
+ */
+package io.camunda.zeebe.engine.state.migration.to_8_3;
+
+import io.camunda.zeebe.engine.state.immutable.ProcessingState;
+import io.camunda.zeebe.engine.state.migration.MigrationTask;
+import io.camunda.zeebe.engine.state.mutable.MutableProcessingState;
+import io.camunda.zeebe.protocol.ZbColumnFamilies;
+
+public final class MultiTenancyMessageStartEventSubscriptionStateMigration
+    implements MigrationTask {
+
+  @Override
+  public String getIdentifier() {
+    return getClass().getSimpleName();
+  }
+
+  @Override
+  public boolean needsToRun(final ProcessingState processingState) {
+    return hasOpenMessageStartEventSubscriptionsInDeprecatedCFs(processingState);
+  }
+
+  @Override
+  public void runMigration(final MutableProcessingState processingState) {
+    final var migrationState = processingState.getMigrationState();
+    migrationState.migrateMessageStartEventSubscriptionForMultiTenancy();
+  }
+
+  private static boolean hasOpenMessageStartEventSubscriptionsInDeprecatedCFs(
+      final ProcessingState processingState) {
+    return !processingState.isEmpty(
+            ZbColumnFamilies.DEPRECATED_MESSAGE_START_EVENT_SUBSCRIPTION_BY_NAME_AND_KEY)
+        || !processingState.isEmpty(
+            ZbColumnFamilies.DEPRECATED_MESSAGE_START_EVENT_SUBSCRIPTION_BY_KEY_AND_NAME);
+  }
+}

engine/src/main/java/io/camunda/zeebe/engine/state/migration/to_8_3/MultiTenancyMigration.java

@@ -44,7 +44,6 @@ public boolean needsToRun(final ProcessingState processingState) {
   @Override
   public void runMigration(final MutableProcessingState processingState) {
     final var migrationState = processingState.getMigrationState();
-    migrationState.migrateMessageStartEventSubscriptionForMultiTenancy();
     migrationState.migrateMessageEventSubscriptionForMultiTenancy();
     migrationState.migrateProcessMessageSubscriptionForMultiTenancy();
     migrationState.migrateJobStateForMultiTenancy();

engine/src/test/java/io/camunda/zeebe/engine/state/migration/to_8_3/MultiTenancyMigrationTest.java

@@ -648,6 +648,9 @@ void shouldMigrateMessagesColumnFamily() {
   @ExtendWith(ProcessingStateExtension.class)
   class MigrateMessageStartEventSubscriptionStateForMultiTenancyTest {
 
+    final MultiTenancyMessageStartEventSubscriptionStateMigration sut =
+        new MultiTenancyMessageStartEventSubscriptionStateMigration();
+
     private ZeebeDb<ZbColumnFamilies> zeebeDb;
     private MutableProcessingState processingState;
     private TransactionContext transactionContext;