diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index b89e10c0e3c..3800184751b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -968,6 +968,9 @@ Release 2.8.0 - UNRELEASED
     HDFS-8546. Prune cached replicas from DatanodeDescriptor state on replica
     invalidation. (wang)
 
+    HDFS-8586. Dead Datanode is allocated for write when client is from deadnode
+    (Brahma Reddy Battula via vinayakumarb)
+
 Release 2.7.2 - UNRELEASED
 
   INCOMPATIBLE CHANGES
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyDefault.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyDefault.java
index 21ad01d162b..9023e0a12b3 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyDefault.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyDefault.java
@@ -454,7 +454,8 @@ protected DatanodeStorageInfo chooseLocalStorage(Node localMachine,
       return chooseRandom(NodeBase.ROOT, excludedNodes, blocksize,
           maxNodesPerRack, results, avoidStaleNodes, storageTypes);
     }
-    if (preferLocalNode && localMachine instanceof DatanodeDescriptor) {
+    if (preferLocalNode && localMachine instanceof DatanodeDescriptor
+        && clusterMap.contains(localMachine)) {
       DatanodeDescriptor localDatanode = (DatanodeDescriptor) localMachine;
       // otherwise try local machine first
       if (excludedNodes.add(localMachine)) { // was not in the excluded list
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDeadDatanode.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDeadDatanode.java
index ff70c3f1445..c5262d457b9 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDeadDatanode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDeadDatanode.java
@@ -18,9 +18,11 @@
 package org.apache.hadoop.hdfs.server.namenode;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.fail;
 
 import java.io.IOException;
+import java.util.HashSet;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -31,6 +33,9 @@
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.protocol.BlockListAsLongs;
+import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
+import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeManager;
+import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeStorageInfo;
 import org.apache.hadoop.hdfs.server.datanode.DataNode;
 import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils;
 import org.apache.hadoop.hdfs.server.protocol.BlockReportContext;
@@ -43,6 +48,7 @@
 import org.apache.hadoop.hdfs.server.protocol.StorageBlockReport;
 import org.apache.hadoop.hdfs.server.protocol.StorageReceivedDeletedBlocks;
 import org.apache.hadoop.hdfs.server.protocol.StorageReport;
+import org.apache.hadoop.net.Node;
 import org.junit.After;
 import org.junit.Test;
 
@@ -126,4 +132,40 @@ public void testDeadDatanode() throws Exception {
     assertEquals(cmd[0].getAction(), RegisterCommand.REGISTER
         .getAction());
   }
+
+  @Test
+  public void testDeadNodeAsBlockTarget() throws Exception {
+    Configuration conf = new HdfsConfiguration();
+    conf.setInt(DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY, 500);
+    conf.setLong(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, 1L);
+    cluster = new MiniDFSCluster.Builder(conf).numDataNodes(3).build();
+    cluster.waitActive();
+
+    String poolId = cluster.getNamesystem().getBlockPoolId();
+    // wait for datanode to be marked live
+    DataNode dn = cluster.getDataNodes().get(0);
+    DatanodeRegistration reg = DataNodeTestUtils.getDNRegistrationForBP(cluster
+        .getDataNodes().get(0), poolId);
+    // Get the updated datanode descriptor
+    BlockManager bm = cluster.getNamesystem().getBlockManager();
+    DatanodeManager dm = bm.getDatanodeManager();
+    Node clientNode = dm.getDatanode(reg);
+
+    DFSTestUtil.waitForDatanodeState(cluster, reg.getDatanodeUuid(), true,
+        20000);
+
+    // Shutdown and wait for datanode to be marked dead
+    dn.shutdown();
+    DFSTestUtil.waitForDatanodeState(cluster, reg.getDatanodeUuid(), false,
+        20000);
+    // Get the updated datanode descriptor available in DNM
+    // choose the targets, but local node should not get selected as this is not
+    // part of the cluster anymore
+    DatanodeStorageInfo[] results = bm.chooseTarget4NewBlock("/hello", 3,
+        clientNode, new HashSet<Node>(), 256 * 1024 * 1024L, null, (byte) 7);
+    for (DatanodeStorageInfo datanodeStorageInfo : results) {
+      assertFalse("Dead node should not be choosen", datanodeStorageInfo
+          .getDatanodeDescriptor().equals(clientNode));
+    }
+  }
 }