Fix: backfill historical gaps in SERVER_CHANGE_LOG during migrate (#27703)

processNativeMigrations used sameOrHigherMajorMinor to gate unexecuted
migrations by the current max's major.minor. That gate silently drops any
version recorded on disk but missing from the ledger whose major.minor is
lower than the current max — for example a historical gap at 1.5.15 when
the dump was last migrated through 1.9.x, or a backported 1.11.11 landing
after 1.12.1 was already cut.

The containment check against SERVER_CHANGE_LOG is sufficient on its own:
any version already executed is skipped, any version not executed is
planned. Maintaining the reprocessing-on-maxVer path preserves the
continuous-release behavior introduced in #26571 (new SQL appended to the
current version's schemaChanges.sql is picked up via checksum filter in
SERVER_MIGRATION_SQL_LOGS).

This also brings processNativeMigrations in line with
processExtensionMigrations, which already uses the same containment-only
filter. See also #26592 which applied the equivalent simplification on
the 1.12.x release line.

Tests:
- Updated three tests that encoded the old exclusion behavior to assert
  backports and historical gaps now get planned.
- Added a historical-gap scenario (1.5.15 below executed max 1.10.5).
- Removed sameOrHigherMajorMinorComparisons (method is gone).

Co-authored-by: Mohit Yadav <105265192+mohityadav766@users.noreply.github.com>

Author: sonika-shah

openmetadata-service/src/main/java/org/openmetadata/service/migration/api/MigrationWorkflow.java

@@ -263,13 +263,6 @@ private static int[] parseVersion(String version) {
     return numbers;
   }
 
-  static boolean sameOrHigherMajorMinor(String version, String maxVersion) {
-    int[] v = parseVersion(version);
-    int[] max = parseVersion(maxVersion);
-    if (v[0] != max[0]) return v[0] > max[0];
-    return v[1] >= max[1];
-  }
-
   // Package-private for testing
   List<MigrationFile> resolveApplyMigrations(List<MigrationFile> availableMigrations) {
     LOG.debug("Filtering Server Migrations");
@@ -321,8 +314,7 @@ private List<MigrationFile> processNativeMigrations(
     for (MigrationFile migration : nativeMigrations) {
       if (migration.version.equals(maxVer)) {
         result.add(migration.copyWithReprocessing(true));
-      } else if (!executedMigrations.contains(migration.version)
-          && sameOrHigherMajorMinor(migration.version, maxVer)) {
+      } else if (!executedMigrations.contains(migration.version)) {
         result.add(migration.copyWithReprocessing(false));
       }
     }

openmetadata-service/src/test/java/org/openmetadata/service/migration/api/MigrationWorkflowTest.java

@@ -410,7 +410,7 @@ void runMigrationWorkflowsExecutesDataMigrationForReprocessing() throws Exceptio
   }
 
   @Test
-  void getMigrationsToApplyOlderMinorBackfillsExcludedWhenOnHigherMinor() throws Exception {
+  void getMigrationsToApplyBackportedLowerMinorVersionsAreIncluded() throws Exception {
     List<String> executedMigrations = List.of("1.11.10", "1.12.0", "1.12.1");
     List<MigrationFile> availableMigrations =
         List.of(
@@ -429,9 +429,31 @@ void getMigrationsToApplyOlderMinorBackfillsExcludedWhenOnHigherMinor() throws E
         workflow.getMigrationsToApply(executedMigrations, availableMigrations);
 
     List<String> versions = result.stream().map(m -> m.version).toList();
-    assertEquals(List.of("1.12.1", "1.12.2"), versions);
-    assertTrue(result.get(0).isReprocessing());
-    assertFalse(result.get(1).isReprocessing());
+    assertEquals(List.of("1.11.11", "1.11.12", "1.12.1", "1.12.2"), versions);
+    assertFalse(
+        result.stream()
+            .filter(m -> m.version.equals("1.11.11"))
+            .findFirst()
+            .orElseThrow()
+            .isReprocessing());
+    assertFalse(
+        result.stream()
+            .filter(m -> m.version.equals("1.11.12"))
+            .findFirst()
+            .orElseThrow()
+            .isReprocessing());
+    assertTrue(
+        result.stream()
+            .filter(m -> m.version.equals("1.12.1"))
+            .findFirst()
+            .orElseThrow()
+            .isReprocessing());
+    assertFalse(
+        result.stream()
+            .filter(m -> m.version.equals("1.12.2"))
+            .findFirst()
+            .orElseThrow()
+            .isReprocessing());
   }
 
   @Test
@@ -458,7 +480,7 @@ void getMigrationsToApplyCollateVersionsAreIncluded() throws Exception {
     List<String> extensionVersions =
         result.stream().filter(m -> m.isExtension).map(m -> m.version).toList();
 
-    assertEquals(List.of("1.12.1", "1.12.2"), nativeVersions);
+    assertEquals(List.of("1.11.11", "1.12.1", "1.12.2"), nativeVersions);
     assertEquals(List.of("1.12.1-collate"), extensionVersions);
     assertTrue(result.stream().anyMatch(m -> m.version.equals("1.12.1") && m.isReprocessing()));
     assertTrue(
@@ -606,7 +628,7 @@ void getMigrationsToApplyAllMigrationsAlreadyExecuted() throws Exception {
   }
 
   @Test
-  void getMigrationsToApplyMultipleBackportedMinorVersionsExcluded() throws Exception {
+  void getMigrationsToApplyMultipleBackportedMinorVersionsAreIncluded() throws Exception {
     List<String> executedMigrations = List.of("1.10.5", "1.11.0", "1.12.0", "1.12.1");
     List<MigrationFile> availableMigrations =
         List.of(
@@ -627,8 +649,50 @@ void getMigrationsToApplyMultipleBackportedMinorVersionsExcluded() throws Except
         workflow.getMigrationsToApply(executedMigrations, availableMigrations);
 
     List<String> versions = result.stream().map(m -> m.version).toList();
-    assertEquals(List.of("1.12.1", "1.12.2"), versions);
-    assertTrue(result.get(0).isReprocessing());
+    assertEquals(List.of("1.10.6", "1.11.1", "1.11.1-collate", "1.12.1", "1.12.2"), versions);
+    assertTrue(
+        result.stream()
+            .filter(m -> m.version.equals("1.12.1"))
+            .findFirst()
+            .orElseThrow()
+            .isReprocessing());
+  }
+
+  @Test
+  void getMigrationsToApplyHistoricalGapBelowMaxIsBackfilled() throws Exception {
+    List<String> executedMigrations =
+        List.of("1.5.0", "1.5.11", "1.9.0", "1.9.1", "1.10.0", "1.10.5");
+    List<MigrationFile> availableMigrations =
+        List.of(
+            createMigrationFile("1.5.0", false),
+            createMigrationFile("1.5.11", false),
+            createMigrationFile("1.5.15", false),
+            createMigrationFile("1.9.0", false),
+            createMigrationFile("1.9.1", false),
+            createMigrationFile("1.10.0", false),
+            createMigrationFile("1.10.5", false));
+
+    MigrationWorkflow workflow =
+        new MigrationWorkflow(
+            jdbi, tempDir.toString(), ConnectionType.MYSQL, null, null, config, false);
+
+    List<MigrationFile> result =
+        workflow.getMigrationsToApply(executedMigrations, availableMigrations);
+
+    List<String> versions = result.stream().map(m -> m.version).toList();
+    assertEquals(List.of("1.5.15", "1.10.5"), versions);
+    assertFalse(
+        result.stream()
+            .filter(m -> m.version.equals("1.5.15"))
+            .findFirst()
+            .orElseThrow()
+            .isReprocessing());
+    assertTrue(
+        result.stream()
+            .filter(m -> m.version.equals("1.10.5"))
+            .findFirst()
+            .orElseThrow()
+            .isReprocessing());
   }
 
   @Test
@@ -696,19 +760,6 @@ void resolveApplyMigrationsAllExecutedNoDuplicates() throws Exception {
     assertTrue(result.get(0).isReprocessing());
   }
 
-  @Test
-  void sameOrHigherMajorMinorComparisons() {
-    assertTrue(MigrationWorkflow.sameOrHigherMajorMinor("1.12.0", "1.12.1"));
-    assertTrue(MigrationWorkflow.sameOrHigherMajorMinor("1.12.5", "1.12.1"));
-    assertTrue(MigrationWorkflow.sameOrHigherMajorMinor("1.13.0", "1.12.1"));
-    assertTrue(MigrationWorkflow.sameOrHigherMajorMinor("2.0.0", "1.12.1"));
-    assertFalse(MigrationWorkflow.sameOrHigherMajorMinor("1.11.15", "1.12.1"));
-    assertFalse(MigrationWorkflow.sameOrHigherMajorMinor("1.10.6", "1.12.1"));
-    assertFalse(MigrationWorkflow.sameOrHigherMajorMinor("0.13.0", "1.12.1"));
-    assertTrue(MigrationWorkflow.sameOrHigherMajorMinor("1.12.1-collate", "1.12.1"));
-    assertFalse(MigrationWorkflow.sameOrHigherMajorMinor("1.11.1-collate", "1.12.1"));
-  }
-
   private void mockContext(
       MigrationWorkflowContext contextMock, org.mockito.MockedConstruction.Context context) {
     HashMap<String, MigrationContext> contexts = new HashMap<>();