HDDS-11470. Report OM snapshot installation failures to Ratis

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMRatisSnapshots.java

@@ -29,6 +29,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.File;
@@ -626,9 +627,10 @@ public void testInstallSnapshotFailedBackupRestoresDbDir() throws Exception {
         "Simulated backup move failure for test"));
     followerOM.setExitManagerForTesting(new DummyExitManager());
     try {
-      TermIndex termIndex = followerOM.installCheckpoint(
-          leaderOMNodeId, leaderCheckpointLocation, leaderCheckpointTrxnInfo);
-      assertNull(termIndex, "Install should have been reported as failed");
+      IOException exception = assertThrows(IOException.class, () ->
+          followerOM.installCheckpoint(leaderOMNodeId, leaderCheckpointLocation,
+              leaderCheckpointTrxnInfo));
+      assertThat(exception).hasMessageContaining("Cannot replace DB");
 
       // Everything present before the aborted install must still be present.
       assertThat(topLevelNames(followerMetaDir)).containsAll(namesBefore);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java

@@ -4260,7 +4260,7 @@ public synchronized TermIndex installSnapshotFromLeader(String leaderId) throws
       }
       termIndex = installCheckpoint(leaderId, checkpointLocation);
     } catch (Exception ex) {
-      LOG.error("Failed to install snapshot from Leader OM.", ex);
+      throw new IOException("Failed to install snapshot from Leader " + leaderId, ex);
     } finally {
       cleanupCheckpoint(omDBCheckpoint);
     }
@@ -4317,6 +4317,7 @@ TermIndex installCheckpoint(String leaderId, Path checkpointLocation,
     long startTime = Time.monotonicNow();
     File oldDBLocation = metadataManager.getStore().getDbLocation();
     Path omDbPath = Paths.get(checkpointLocation.toString(), OM_DB_NAME);
+    IOException installFailure = null;
     try {
       // Stop Background services
       keyManager.stop();
@@ -4328,20 +4329,16 @@ TermIndex installCheckpoint(String leaderId, Path checkpointLocation,
       // pending transactions in the buffer, they are discarded.
       omRatisServer.getOmStateMachine().pause();
     } catch (Exception e) {
-      LOG.error("Failed to stop/ pause the services. Cannot proceed with " +
-          "installing the new checkpoint.");
       // Stop the checkpoint install process and restart the services.
       keyManager.start(configuration);
       startSecretManagerIfNecessary();
       startTrashEmptier(configuration);
-      throw e;
+      throw newInstallCheckpointException(checkpointTrxnInfo, "stop/pause services", e);
     }
-
     File dbBackup = null;
     TermIndex termIndex = omRatisServer.getLastAppliedTermIndex();
     long term = termIndex.getTerm();
     long lastAppliedIndex = termIndex.getIndex();
-
     // Check if current applied log index is smaller than the downloaded
     // checkpoint transaction index. If yes, proceed by stopping the ratis
     // server so that the OM state can be re-initialized. If no then do not
@@ -4383,6 +4380,7 @@ TermIndex installCheckpoint(String leaderId, Path checkpointLocation,
         LOG.error("Failed to install Snapshot from {} as OM failed to replace" +
             " DB with downloaded checkpoint. Reloading old OM state.",
             leaderId, e);
+        installFailure = newInstallCheckpointException(checkpointTrxnInfo, "replace DB", e);
       }
     } else {
       LOG.warn("Cannot proceed with InstallSnapshot as OM is at TermIndex {} " +
@@ -4449,6 +4447,8 @@ TermIndex installCheckpoint(String leaderId, Path checkpointLocation,
           dbBackup, e);
     }
 
+    throwIfInstallCheckpointFailed(installFailure);
+
     if (lastAppliedIndex != checkpointTrxnInfo.getTransactionIndex()) {
       // Install Snapshot failed and old state was reloaded. Return null to
       // Ratis to indicate that installation failed.
@@ -4464,6 +4464,18 @@ TermIndex installCheckpoint(String leaderId, Path checkpointLocation,
     return newTermIndex;
   }
 
+  private static IOException newInstallCheckpointException(TransactionInfo checkpointTrxnInfo,
+      String operation, Exception cause) {
+    return new IOException("Failed to install checkpoint " + checkpointTrxnInfo
+        + ": Cannot " + operation + '.', cause);
+  }
+
+  private static void throwIfInstallCheckpointFailed(IOException installFailure) throws IOException {
+    if (installFailure != null) {
+      throw installFailure;
+    }
+  }
+
   private void buildDBCheckpointInstallAuditLog(String leaderId, long term, long lastAppliedIndex) {
     Map<String, String> auditMap = new LinkedHashMap<>();
     auditMap.put(AUDIT_PARAM_LEADER_ID, leaderId);

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java

@@ -829,6 +829,38 @@ public void testNotifyConfigurationChanged() {
 
   // --- notifySnapshotInstalled tests ---
 
+  @Test
+  public void testNotifyInstallSnapshotFromLeaderSuccess() throws Exception {
+    TermIndex termIndex = TermIndex.valueOf(1, 10);
+    when(om.installSnapshotFromLeader("leader-om")).thenReturn(termIndex);
+
+    CompletableFuture<TermIndex> future = sm.notifyInstallSnapshotFromLeader(
+        roleInfoWithLeader("leader-om"), TermIndex.valueOf(1, 11));
+
+    assertEquals(termIndex, future.get());
+  }
+
+  @Test
+  public void testNotifyInstallSnapshotFromLeaderNullResult() throws Exception {
+    CompletableFuture<TermIndex> future = sm.notifyInstallSnapshotFromLeader(
+        roleInfoWithLeader("leader-om"), TermIndex.valueOf(1, 11));
+
+    assertNull(future.get());
+  }
+
+  @Test
+  public void testNotifyInstallSnapshotFromLeaderFailure() throws Exception {
+    IOException failure = new IOException("Failed to install checkpoint");
+    doThrow(failure).when(om).installSnapshotFromLeader("leader-om");
+
+    CompletableFuture<TermIndex> future = sm.notifyInstallSnapshotFromLeader(
+        roleInfoWithLeader("leader-om"), TermIndex.valueOf(1, 11));
+
+    ExecutionException exception = assertThrows(ExecutionException.class, future::get);
+
+    assertSame(failure, exception.getCause());
+  }
+
   @Test
   public void testNotifySnapshotInstalledSuccess() {
     RaftPeer localPeer = RaftPeer.newBuilder()
@@ -1044,6 +1076,15 @@ private RaftClientRequest buildClientRequest(
         .build();
   }
 
+  private RaftProtos.RoleInfoProto roleInfoWithLeader(String leaderId) {
+    return RaftProtos.RoleInfoProto.newBuilder()
+        .setFollowerInfo(RaftProtos.FollowerInfoProto.newBuilder()
+            .setLeaderInfo(RaftProtos.ServerRpcProto.newBuilder()
+                .setId(RaftProtos.RaftPeerProto.newBuilder()
+                    .setId(ByteString.copyFromUtf8(leaderId)))))
+        .build();
+  }
+
   @Test
   public void testRatisEventsRecording() {
     OzoneConfiguration conf = new OzoneConfiguration();