diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenIdentifier.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenIdentifier.java
index 76d1a170a40..1d395ec24ac 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenIdentifier.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenIdentifier.java
@@ -248,31 +248,34 @@ public class BlockTokenIdentifier extends TokenIdentifier {
       modes.add(WritableUtils.readEnum(in, AccessMode.class));
     }
 
-    length = WritableUtils.readVInt(in);
-    StorageType[] readStorageTypes = new StorageType[length];
-    for (int i = 0; i < length; i++) {
-      readStorageTypes[i] = WritableUtils.readEnum(in, StorageType.class);
-    }
-    storageTypes = readStorageTypes;
-
-    length = WritableUtils.readVInt(in);
-    String[] readStorageIds = new String[length];
-    for (int i = 0; i < length; i++) {
-      readStorageIds[i] = WritableUtils.readString(in);
-    }
-    storageIds = readStorageIds;
-
-    useProto = false;
-
     try {
+      length = WritableUtils.readVInt(in);
+      StorageType[] readStorageTypes = new StorageType[length];
+      for (int i = 0; i < length; i++) {
+        readStorageTypes[i] = WritableUtils.readEnum(in, StorageType.class);
+      }
+      storageTypes = readStorageTypes;
+
+      length = WritableUtils.readVInt(in);
+      String[] readStorageIds = new String[length];
+      for (int i = 0; i < length; i++) {
+        readStorageIds[i] = WritableUtils.readString(in);
+      }
+      storageIds = readStorageIds;
+
       int handshakeMsgLen = WritableUtils.readVInt(in);
       if (handshakeMsgLen != 0) {
         handshakeMsg = new byte[handshakeMsgLen];
         in.readFully(handshakeMsg);
       }
     } catch (EOFException eof) {
-
+      // If the NameNode is on a version before HDFS-6708 and HDFS-9807, then
+      // the block token won't have storage types or storage IDs. For backward
+      // compatibility, swallow the EOF that we get when we try to read those
+      // fields. Same for the handshake secret field from HDFS-14611.
     }
+
+    useProto = false;
   }
 
   @VisibleForTesting
@@ -332,13 +335,17 @@ public class BlockTokenIdentifier extends TokenIdentifier {
     for (AccessMode aMode : modes) {
       WritableUtils.writeEnum(out, aMode);
     }
-    WritableUtils.writeVInt(out, storageTypes.length);
-    for (StorageType type: storageTypes){
-      WritableUtils.writeEnum(out, type);
+    if (storageTypes != null) {
+      WritableUtils.writeVInt(out, storageTypes.length);
+      for (StorageType type : storageTypes) {
+        WritableUtils.writeEnum(out, type);
+      }
     }
-    WritableUtils.writeVInt(out, storageIds.length);
-    for (String id: storageIds) {
-      WritableUtils.writeString(out, id);
+    if (storageIds != null) {
+      WritableUtils.writeVInt(out, storageIds.length);
+      for (String id : storageIds) {
+        WritableUtils.writeString(out, id);
+      }
     }
     if (handshakeMsg != null && handshakeMsg.length > 0) {
       WritableUtils.writeVInt(out, handshakeMsg.length);
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java
index fe5ee65393a..7961150ee00 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java
@@ -42,6 +42,7 @@ import java.util.GregorianCalendar;
 import java.util.Set;
 
 import org.mockito.Mockito;
+import org.apache.commons.lang3.reflect.FieldUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.conf.Configuration;
@@ -614,6 +615,24 @@ public class TestBlockToken {
     readToken.readFields(dib);
   }
 
+  /**
+   * If the NameNode predates HDFS-6708 and HDFS-9807, then the LocatedBlocks
+   * that it returns to the client will have block tokens that don't include
+   * the storage types or storage IDs. Simulate this by setting the storage
+   * type and storage ID to null to test backwards compatibility.
+   */
+  @Test
+  public void testLegacyBlockTokenWithoutStorages() throws IOException,
+          IllegalAccessException {
+    BlockTokenIdentifier identifier = new BlockTokenIdentifier("user",
+            "blockpool", 123,
+            EnumSet.allOf(BlockTokenIdentifier.AccessMode.class), null, null,
+            false);
+    FieldUtils.writeField(identifier, "storageTypes", null, true);
+    FieldUtils.writeField(identifier, "storageIds", null, true);
+    testCraftedBlockTokenIdentifier(identifier, false, false, false);
+  }
+
   @Test
   public void testProtobufBlockTokenBytesIsProtobuf() throws IOException {
     final boolean useProto = true;
@@ -659,13 +678,17 @@ public class TestBlockToken {
     assertEquals(protobufToken, readToken);
   }
 
-  private void testCraftedProtobufBlockTokenIdentifier(
+  private void testCraftedBlockTokenIdentifier(
       BlockTokenIdentifier identifier, boolean expectIOE,
-      boolean expectRTE) throws IOException {
+      boolean expectRTE, boolean isProtobuf) throws IOException {
     DataOutputBuffer dob = new DataOutputBuffer(4096);
     DataInputBuffer dib = new DataInputBuffer();
 
-    identifier.writeProtobuf(dob);
+    if (isProtobuf) {
+      identifier.writeProtobuf(dob);
+    } else {
+      identifier.writeLegacy(dob);
+    }
     byte[] identBytes = Arrays.copyOf(dob.getData(), dob.getLength());
 
     BlockTokenIdentifier legacyToken = new BlockTokenIdentifier();
@@ -688,22 +711,23 @@ public class TestBlockToken {
       invalidLegacyMessage = true;
     }
 
-    assertTrue(invalidLegacyMessage);
+    if (isProtobuf) {
+      assertTrue(invalidLegacyMessage);
 
-    dib.reset(identBytes, identBytes.length);
-    protobufToken.readFieldsProtobuf(dib);
-
-    dib.reset(identBytes, identBytes.length);
-    readToken.readFieldsProtobuf(dib);
-    assertEquals(protobufToken, readToken);
-    assertEquals(identifier, readToken);
+      dib.reset(identBytes, identBytes.length);
+      protobufToken.readFieldsProtobuf(dib);
+      dib.reset(identBytes, identBytes.length);
+      readToken.readFields(dib);
+      assertEquals(identifier, readToken);
+      assertEquals(protobufToken, readToken);
+    }
   }
 
   @Test
   public void testEmptyProtobufBlockTokenBytesIsProtobuf() throws IOException {
     // Empty BlockTokenIdentifiers throw IOException
     BlockTokenIdentifier identifier = new BlockTokenIdentifier();
-    testCraftedProtobufBlockTokenIdentifier(identifier, true, false);
+    testCraftedBlockTokenIdentifier(identifier, true, false, true);
   }
 
   @Test
@@ -724,10 +748,10 @@ public class TestBlockToken {
     datetime = ((datetime / 1000) * 1000); // strip milliseconds.
     datetime = datetime + 71; // 2017-02-09 00:12:35,071+0100
     identifier.setExpiryDate(datetime);
-    testCraftedProtobufBlockTokenIdentifier(identifier, false, true);
+    testCraftedBlockTokenIdentifier(identifier, false, true, true);
     datetime += 1; // 2017-02-09 00:12:35,072+0100
     identifier.setExpiryDate(datetime);
-    testCraftedProtobufBlockTokenIdentifier(identifier, true, false);
+    testCraftedBlockTokenIdentifier(identifier, true, false, true);
   }
 
   private BlockTokenIdentifier writeAndReadBlockToken(