HDDS-10408. NPE causes OM crash in Snapshot Purge request (apache#6250)

(cherry picked from commit 6dfd7d4)

Author: aswinshakil

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/snapshot/OMSnapshotPurgeRequest.java

@@ -34,6 +34,8 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SnapshotPurgeRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.HashMap;
@@ -48,6 +50,8 @@
  */
 public class OMSnapshotPurgeRequest extends OMClientRequest {
 
+  private static final Logger LOG = LoggerFactory.getLogger(OMSnapshotPurgeRequest.class);
+
   public OMSnapshotPurgeRequest(OMRequest omRequest) {
     super(omRequest);
   }
@@ -94,6 +98,13 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
         SnapshotInfo fromSnapshot = omMetadataManager.getSnapshotInfoTable()
             .get(snapTableKey);
 
+        if (fromSnapshot == null) {
+          // Snapshot may have been purged in the previous iteration of SnapshotDeletingService.
+          LOG.warn("The snapshot {} is not longer in snapshot table, It maybe removed in the previous " +
+                  "Snapshot purge request.", snapTableKey);
+          continue;
+        }
+
         SnapshotInfo nextSnapshot = SnapshotUtils
             .getNextActiveSnapshot(fromSnapshot,
                 snapshotChainManager, omSnapshotManager);

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

@@ -40,6 +40,7 @@
 import java.util.UUID;
 
 import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
+import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.INVALID_SNAPSHOT_ERROR;
 import static org.apache.hadoop.ozone.om.OmMetadataManagerImpl.DIRECTORY_TABLE;
 import static org.apache.hadoop.ozone.om.OmMetadataManagerImpl.FILE_TABLE;
 import static org.apache.hadoop.ozone.om.OmMetadataManagerImpl.KEY_TABLE;
@@ -148,6 +149,10 @@ public static SnapshotInfo getNextActiveSnapshot(SnapshotInfo snapInfo,
     // If the snapshot is deleted in the previous run, then the in-memory
     // SnapshotChainManager might throw NoSuchElementException as the snapshot
     // is removed in-memory but OMDoubleBuffer has not flushed yet.
+    if (snapInfo == null) {
+      throw new OMException("Snapshot Info is null. Cannot get the next snapshot", INVALID_SNAPSHOT_ERROR);
+    }
+
     try {
       while (chainManager.hasNextPathSnapshot(snapInfo.getSnapshotPath(),
           snapInfo.getSnapshotId())) {