HDFS-15566. NN restart fails after RollingUpgrade from 3.1.3/3.2.1 to 3.3.0. Contributed by Brahma Reddy Battula.

Signed-off-by: Wei-Chiu Chuang <weichiu@apache.org>

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

@@ -3477,17 +3477,30 @@ CreateSnapshotOp setSnapshotMTime(long mTime) {
     void readFields(DataInputStream in, int logVersion) throws IOException {
       snapshotRoot = FSImageSerialization.readString(in);
       snapshotName = FSImageSerialization.readString(in);
-      mtime = FSImageSerialization.readLong(in);
+      if (NameNodeLayoutVersion
-      
+          .supports(NameNodeLayoutVersion.Feature.SNAPSHOT_MODIFICATION_TIME,
+              logVersion)) {
+        mtime = FSImageSerialization.readLong(in);
+      }
       // read RPC ids if necessary
       readRpcIds(in, logVersion);
     }
 
     @Override
     public void writeFields(DataOutputStream out) throws IOException {
+      throw new IOException("Unsupported without logversion");
+    }
+
+    @Override
+    public void writeFields(DataOutputStream out, int logVersion)
+        throws IOException {
       FSImageSerialization.writeString(snapshotRoot, out);
       FSImageSerialization.writeString(snapshotName, out);
-      FSImageSerialization.writeLong(mtime, out);
+      if (NameNodeLayoutVersion
+          .supports(NameNodeLayoutVersion.Feature.SNAPSHOT_MODIFICATION_TIME,
+              logVersion)) {
+        FSImageSerialization.writeLong(mtime, out);
+      }
       writeRpcIds(rpcClientId, rpcCallId, out);
     }
 
@@ -3569,17 +3582,30 @@ DeleteSnapshotOp setSnapshotMTime(long mTime) {
     void readFields(DataInputStream in, int logVersion) throws IOException {
       snapshotRoot = FSImageSerialization.readString(in);
       snapshotName = FSImageSerialization.readString(in);
-      mtime = FSImageSerialization.readLong(in);
+      if (NameNodeLayoutVersion
-      
+          .supports(NameNodeLayoutVersion.Feature.SNAPSHOT_MODIFICATION_TIME,
+              logVersion)) {
+        mtime = FSImageSerialization.readLong(in);
+      }
       // read RPC ids if necessary
       readRpcIds(in, logVersion);
     }
 
     @Override
     public void writeFields(DataOutputStream out) throws IOException {
+      throw new IOException("Unsupported without logversion");
+    }
+
+    @Override
+    public void writeFields(DataOutputStream out, int logVersion)
+        throws IOException {
       FSImageSerialization.writeString(snapshotRoot, out);
       FSImageSerialization.writeString(snapshotName, out);
-      FSImageSerialization.writeLong(mtime, out);
+      if (NameNodeLayoutVersion
+          .supports(NameNodeLayoutVersion.Feature.SNAPSHOT_MODIFICATION_TIME,
+              logVersion)) {
+        FSImageSerialization.writeLong(mtime, out);
+      }
       writeRpcIds(rpcClientId, rpcCallId, out);
     }
 
@@ -3670,19 +3696,31 @@ void readFields(DataInputStream in, int logVersion) throws IOException {
       snapshotRoot = FSImageSerialization.readString(in);
       snapshotOldName = FSImageSerialization.readString(in);
       snapshotNewName = FSImageSerialization.readString(in);
-      mtime = FSImageSerialization.readLong(in);
+      if (NameNodeLayoutVersion
-      
+          .supports(NameNodeLayoutVersion.Feature.SNAPSHOT_MODIFICATION_TIME,
+              logVersion)) {
+        mtime = FSImageSerialization.readLong(in);
+      }
       // read RPC ids if necessary
       readRpcIds(in, logVersion);
     }
 
     @Override
     public void writeFields(DataOutputStream out) throws IOException {
+      throw new IOException("Unsupported without logversion");
+    }
+
+    @Override
+    public void writeFields(DataOutputStream out, int logVersion)
+        throws IOException {
       FSImageSerialization.writeString(snapshotRoot, out);
       FSImageSerialization.writeString(snapshotOldName, out);
       FSImageSerialization.writeString(snapshotNewName, out);
-      FSImageSerialization.writeLong(mtime, out);
+      if (NameNodeLayoutVersion
-      
+          .supports(NameNodeLayoutVersion.Feature.SNAPSHOT_MODIFICATION_TIME,
+              logVersion)) {
+        FSImageSerialization.writeLong(mtime, out);
+      }
       writeRpcIds(rpcClientId, rpcCallId, out);
     }
 

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

@@ -89,7 +89,8 @@ public enum Feature implements LayoutFeature {
     APPEND_NEW_BLOCK(-62, -61, "Support appending to new block"),
     QUOTA_BY_STORAGE_TYPE(-63, -61, "Support quota for specific storage types"),
     ERASURE_CODING(-64, -61, "Support erasure coding"),
-    EXPANDED_STRING_TABLE(-65, -61, "Support expanded string table in fsimage");
+    EXPANDED_STRING_TABLE(-65, -61, "Support expanded string table in fsimage"),
+    SNAPSHOT_MODIFICATION_TIME(-66, -61, "Support modification time for snapshot");
 
     private final FeatureInfo info;
 

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/TestLayoutVersion.java

@@ -128,7 +128,8 @@ public void testNameNodeFeatureMinimumCompatibleLayoutVersions() {
         NameNodeLayoutVersion.Feature.APPEND_NEW_BLOCK,
         NameNodeLayoutVersion.Feature.QUOTA_BY_STORAGE_TYPE,
         NameNodeLayoutVersion.Feature.ERASURE_CODING,
-        NameNodeLayoutVersion.Feature.EXPANDED_STRING_TABLE);
+        NameNodeLayoutVersion.Feature.EXPANDED_STRING_TABLE,
+        NameNodeLayoutVersion.Feature.SNAPSHOT_MODIFICATION_TIME);
     for (LayoutFeature f : compatibleFeatures) {
       assertEquals(String.format("Expected minimum compatible layout version " +
           "%d for feature %s.", baseLV, f), baseLV,