HDDS-16462. Recon full snapshot onto an earlier OM lineage plus crash permanently serves stale derived data.

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java

@@ -278,20 +278,7 @@ public void start() {
     ReconTaskStatusUpdater fullSnapshotReconTaskUpdater = taskStatusUpdaterManager.getTaskStatusUpdater(
         OmSnapshotTaskName.OmSnapshotRequest.name());
 
-    Map<String, ReconOmTask> reconOmTaskMap = reconTaskController.getRegisteredTasks()
-        .entrySet()
-        .stream()
-        .filter(entry -> {
-          String taskName = entry.getKey();
-          ReconTaskStatusUpdater taskStatusUpdater = taskStatusUpdaterManager.getTaskStatusUpdater(taskName);
-
-          return !taskName.equals(OmSnapshotTaskName.OmDeltaRequest.name())  // Condition 1
-              && !taskName.equals(OmSnapshotTaskName.OmSnapshotRequest.name())  // Condition 2
-              &&
-              taskStatusUpdater.getLastUpdatedSeqNumber().compareTo(
-                  deltaTaskStatusUpdater.getLastUpdatedSeqNumber()) < 0; // Condition 3
-        })
-        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));  // Collect into desired Map
+    Map<String, ReconOmTask> reconOmTaskMap = getTasksToReprocessOnStartup(deltaTaskStatusUpdater);
     if (!reconOmTaskMap.isEmpty() && omMetadataManager.getStore() == null) {
       // Fresh start (or the local OM snapshot DB is missing) while stale task
       // status rows still exist in the Recon SQL DB. There is no local OM DB to
@@ -333,6 +320,34 @@ public void start() {
     LOG.info("Ozone Manager Service Provider is started.");
   }
 
+  /**
+   * Selects the registered derived OM tasks that must be reprocessed at startup because their
+   * persisted sequence cursor does not match the OmDeltaRequest cursor. A derived task is
+   * reprocessed whenever the two cursors differ, not only when the derived cursor lags. After a
+   * full-snapshot swap onto a different OM lineage (for example a disaster-recovery restore or
+   * re-bootstrap), the freshly installed OM DB copy can carry a sequence number lower than what the
+   * derived task last recorded against the previous lineage. If the async reprocess queued by the
+   * full snapshot is lost to a restart, a strict "less than" check would read the higher stale cursor
+   * as "caught up" and skip the rebuild, leaving the derived tables permanently built against the old
+   * lineage. A new lineage that lands on exactly the same sequence number cannot be detected here.
+   */
+  Map<String, ReconOmTask> getTasksToReprocessOnStartup(ReconTaskStatusUpdater deltaTaskStatusUpdater) {
+    return reconTaskController.getRegisteredTasks()
+        .entrySet()
+        .stream()
+        .filter(entry -> {
+          String taskName = entry.getKey();
+          ReconTaskStatusUpdater taskStatusUpdater = taskStatusUpdaterManager.getTaskStatusUpdater(taskName);
+
+          return !taskName.equals(OmSnapshotTaskName.OmDeltaRequest.name())  // Condition 1
+              && !taskName.equals(OmSnapshotTaskName.OmSnapshotRequest.name())  // Condition 2
+              &&
+              taskStatusUpdater.getLastUpdatedSeqNumber().compareTo(
+                  deltaTaskStatusUpdater.getLastUpdatedSeqNumber()) != 0; // Condition 3
+        })
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));  // Collect into desired Map
+  }
+
   private void startSyncDataFromOM(long initialDelay) {
     long interval = configuration.getTimeDuration(
         OZONE_RECON_OM_SNAPSHOT_TASK_INTERVAL_DELAY,

hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java

@@ -59,7 +59,9 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -93,6 +95,7 @@
 import org.apache.hadoop.ozone.recon.metrics.OzoneManagerSyncMetrics;
 import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
 import org.apache.hadoop.ozone.recon.tasks.OMUpdateEventBatch;
+import org.apache.hadoop.ozone.recon.tasks.ReconOmTask;
 import org.apache.hadoop.ozone.recon.tasks.ReconTaskController;
 import org.apache.hadoop.ozone.recon.tasks.ReconTaskControllerImpl;
 import org.apache.hadoop.ozone.recon.tasks.ReconTaskReInitializationEvent;
@@ -977,6 +980,46 @@ configuration, mock(ReconOMMetadataManager.class), noDbController,
     verify(noDbController, never()).queueReInitializationEvent(any());
   }
 
+  @Test
+  public void testStartupReconciliationReprocessesTaskOnLineageRegression() {
+    ReconTaskController reconTaskController = mock(ReconTaskController.class);
+    ReconTaskStatusUpdaterManager taskStatusUpdaterManager = mock(ReconTaskStatusUpdaterManager.class);
+    ReconOMMetadataManager omMetadataManager = mock(ReconOMMetadataManager.class);
+    OzoneManagerServiceProviderImpl serviceProvider = new OzoneManagerServiceProviderImpl(
+        configuration, omMetadataManager, reconTaskController, new ReconUtils(),
+        ozoneManagerProtocol, reconContext, taskStatusUpdaterManager);
+
+    String derivedTaskName = "ContainerKeyMapperTask";
+    Map<String, ReconOmTask> registeredTasks = new HashMap<>();
+    registeredTasks.put(derivedTaskName, mock(ReconOmTask.class));
+    when(reconTaskController.getRegisteredTasks()).thenReturn(registeredTasks);
+
+    ReconTaskStatusUpdater deltaUpdater = mock(ReconTaskStatusUpdater.class);
+    ReconTaskStatusUpdater derivedUpdater = mock(ReconTaskStatusUpdater.class);
+    when(taskStatusUpdaterManager.getTaskStatusUpdater(derivedTaskName)).thenReturn(derivedUpdater);
+
+    // Lineage regression: a full-snapshot swap installed an OM DB copy whose sequence number (2) is
+    // lower than what the derived task recorded against the previous lineage (5). The stale derived
+    // cursor is numerically ahead of the delta cursor, so a strict "less than" check would treat the
+    // stale derived tables as caught up and skip the rebuild.
+    when(deltaUpdater.getLastUpdatedSeqNumber()).thenReturn(2L);
+    when(derivedUpdater.getLastUpdatedSeqNumber()).thenReturn(5L);
+    assertThat(serviceProvider.getTasksToReprocessOnStartup(deltaUpdater))
+        .as("derived task must be reprocessed when its cursor is ahead of a lower new-lineage delta cursor")
+        .containsKey(derivedTaskName);
+
+    // Normal lag (derived cursor behind delta cursor) is still reprocessed: original behavior preserved.
+    when(derivedUpdater.getLastUpdatedSeqNumber()).thenReturn(1L);
+    assertThat(serviceProvider.getTasksToReprocessOnStartup(deltaUpdater))
+        .containsKey(derivedTaskName);
+
+    // Healthy caught-up state (equal cursors) must NOT trigger a reprocess (no over-triggering).
+    when(derivedUpdater.getLastUpdatedSeqNumber()).thenReturn(2L);
+    assertThat(serviceProvider.getTasksToReprocessOnStartup(deltaUpdater))
+        .as("equal cursors mean the derived tables are current and must not be rebuilt")
+        .isEmpty();
+  }
+
   private <T extends OMMetadataManager> T stopAfterTest(T omMetadataManager) {
     omMetadataManagersToStop.add(omMetadataManager);
     return omMetadataManager;