diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/security/token/delegation/BundledDelegationTokenIdentifier.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/security/token/delegation/BundledDelegationTokenIdentifier.java
index 0a47dfc37e83b..0c010e9cd1ffb 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/security/token/delegation/BundledDelegationTokenIdentifier.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/security/token/delegation/BundledDelegationTokenIdentifier.java
@@ -21,10 +21,34 @@
  */
 public class BundledDelegationTokenIdentifier extends DelegationTokenIdentifier
     implements BundledTokenIdentifier {
+
+  // The current VERSION of the token.
+  private static final byte VERSION = 0;
+
+  // The designated main password.
+  // This will be empty (byte[0]) if deserialization fails to read any content
+  // for the main password.
   private byte[] mainPassword;
+
+  // The length of all inner tokens.
+  // This will be 0 if deserialization fails to read any content for the
+  // inner tokens.
   private int innerTokensLength;
+
+  // The list of all inner tokens including the main token as well.
+  // This will be empty (Token[0]) if deserialization fails to read any content
+  // for the inner tokens.
   private Token[] innerTokens;
 
+  // The version of the BundledDelegationTokenIdentifier found for this token.
+  // This will be set to BASE_VERSION if deserialization fails to read any
+  // content for the tokenVersion.
+  private byte tokenVersion;
+
+  // The original serde implementation of this identifier was un-versioned.
+  // This denotes the default version for un-versioned tokens.
+  private static final byte BASE_VERSION = 0;
+
   public BundledDelegationTokenIdentifier() {
     super();
   }
@@ -48,11 +72,15 @@ public BundledDelegationTokenIdentifier(AbstractDelegationTokenIdentifier mainTo
     // Set password of the main token, used during SASL handshake.
     this.mainPassword = mainPassword;
 
-    // Set information for all bundled tokens.
-    if (tokens != null) {
-      this.innerTokensLength = tokens.length;
-      this.innerTokens = tokens;
+    if (tokens == null) {
+      tokens = new Token[0];
     }
+
+    // Set information for all bundled tokens.
+    this.innerTokensLength = tokens.length;
+    this.innerTokens = tokens;
+
+    this.tokenVersion = VERSION;
   }
 
   @Override
@@ -60,16 +88,16 @@ public void write(DataOutput out) throws IOException {
     super.write(out);
 
     // Serialize bundled token information.
-    if (this.hasInnerTokens()) {
-      WritableUtils.writeCompressedByteArray(out, this.mainPassword);
-      WritableUtils.writeVInt(out, this.innerTokensLength);
-
-      for (Token innerToken : this.innerTokens) {
-        innerToken.decodeIdentifier().write(out);
-        WritableUtils.writeCompressedByteArray(out, innerToken.getPassword());
-        innerToken.getService().write(out);
-      }
+    WritableUtils.writeCompressedByteArray(out, this.mainPassword);
+    WritableUtils.writeVInt(out, this.innerTokensLength);
+
+    for (Token innerToken : this.innerTokens) {
+      innerToken.decodeIdentifier().write(out);
+      WritableUtils.writeCompressedByteArray(out, innerToken.getPassword());
+      innerToken.getService().write(out);
     }
+
+    out.writeByte(tokenVersion);
   }
 
   @Override
@@ -93,9 +121,22 @@ public void readFields(DataInput in) throws IOException {
 
         this.innerTokens[i] = new Token(id.getBytes(), password, HDFS_DELEGATION_KIND, service);
       }
+
+      try {
+        // In the first iteration of the BundledDelegationTokenIdentifier,
+        // there was no versioning, so this field might be missing.
+        this.tokenVersion = in.readByte();
+      } catch (IOException ex) {
+        this.tokenVersion = BASE_VERSION;
+      }
+
     } catch (IOException ex) {
       // This is expected to happen if the tokenIdentifier has no bundled tokens. The error
       // suggests that we reached end of input as there is no more information received.
+      this.mainPassword = new byte[0];
+      this.innerTokensLength = 0;
+      this.innerTokens = new Token[0];
+      this.tokenVersion = BASE_VERSION;
     }
   }
 
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/hdfs/security/token/delegation/TestBundledDelegationTokenIdentifier.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/hdfs/security/token/delegation/TestBundledDelegationTokenIdentifier.java
index 217b6cec766c5..7f1310407ef83 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/hdfs/security/token/delegation/TestBundledDelegationTokenIdentifier.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/hdfs/security/token/delegation/TestBundledDelegationTokenIdentifier.java
@@ -56,6 +56,10 @@ public void testBundledTokenBackwardsCompatible() throws Exception {
     
     // Validate that no bundled information was found during deserialization.
     assertFalse(bundledTokenId2.hasInnerTokens());
+
+    Token[] actualInnerTokens = bundledTokenId2.getInnerTokens();
+    assertNotNull(actualInnerTokens);
+    assertEquals(0, actualInnerTokens.length);
   }
 
   @Test