HDFS-14697. Backport of HDFS-14513. FSImage which is saving should be clean while NameNode shutdown. Contributed by He Xiaoqiao.

(cherry picked from ee1547a8071fbdebb72d409ddfcd46e6c63f82aa)
(cherry picked from 69e3ac5cbb2547416b27dbcfb458099c3a951101)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImage.java

@@ -72,6 +72,7 @@
 import org.apache.hadoop.log.LogThrottlingHelper;
 import org.apache.hadoop.log.LogThrottlingHelper.LogAction;
 import org.apache.hadoop.util.ExitUtil;
+import org.apache.hadoop.util.ShutdownHookManager;
 import org.apache.hadoop.util.Time;
 
 import com.google.common.base.Joiner;
@@ -87,6 +88,11 @@
 public class FSImage implements Closeable {
   public static final Log LOG = LogFactory.getLog(FSImage.class.getName());
 
+  /**
+   * Priority of the FSImageSaver shutdown hook: {@value}.
+   */
+  public static final int SHUTDOWN_HOOK_PRIORITY = 10;
+
   protected FSEditLog editLog = null;
   private boolean isUpgradeFinalized = false;
 
@@ -1035,6 +1041,21 @@ public FSImageSaver(SaveNamespaceContext context, StorageDirectory sd,
 
     @Override
     public void run() {
+      // Deletes checkpoint file in every storage directory when shutdown.
+      Runnable cancelCheckpointFinalizer = new Runnable() {
+        @Override
+        public void run() {
+          try {
+            deleteCancelledCheckpoint(context.getTxId());
+            LOG.info("FSImageSaver clean checkpoint: txid = "
+                + context.getTxId() + " when meet shutdown.");
+          } catch (IOException e) {
+            LOG.error("FSImageSaver cancel checkpoint threw an exception:", e);
+          }
+        }
+      };
+      ShutdownHookManager.get().addShutdownHook(cancelCheckpointFinalizer,
+          SHUTDOWN_HOOK_PRIORITY);
       try {
         saveFSImage(context, sd, nnf);
       } catch (SaveNamespaceCancelledException snce) {
@@ -1044,6 +1065,13 @@ public void run() {
       } catch (Throwable t) {
         LOG.error("Unable to save image for " + sd.getRoot(), t);
         context.reportErrorOnStorageDirectory(sd);
+        try {
+          deleteCancelledCheckpoint(context.getTxId());
+          LOG.info("FSImageSaver clean checkpoint: txid = "
+              + context.getTxId() + " when meet Throwable.");
+        } catch (IOException e) {
+          LOG.error("FSImageSaver cancel checkpoint threw an exception:", e);
+        }
       }
     }