diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/migration/api/MigrationWorkflow.java b/openmetadata-service/src/main/java/org/openmetadata/service/migration/api/MigrationWorkflow.java
index 2a861f9b133a..86ccb4e0d892 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/migration/api/MigrationWorkflow.java
+++ b/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));
       }
     }
diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/migration/api/MigrationWorkflowTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/migration/api/MigrationWorkflowTest.java
index 8b1efa3df7d2..54561b8107bc 100644
--- a/openmetadata-service/src/test/java/org/openmetadata/service/migration/api/MigrationWorkflowTest.java
+++ b/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<>();