|
37 | 37 | import java.util.concurrent.ExecutionException; |
38 | 38 | import java.util.concurrent.atomic.AtomicBoolean; |
39 | 39 | import java.util.regex.Pattern; |
| 40 | +import java.util.EnumSet; |
40 | 41 |
|
41 | 42 | import java.util.function.Supplier; |
42 | 43 | import org.apache.hadoop.thirdparty.com.google.common.collect.Lists; |
|
46 | 47 | import org.apache.hadoop.fs.FSDataOutputStream; |
47 | 48 | import org.apache.hadoop.fs.FileSystem; |
48 | 49 | import org.apache.hadoop.fs.Path; |
| 50 | +import org.apache.hadoop.fs.BatchedRemoteIterator.BatchedEntries; |
49 | 51 | import org.apache.hadoop.hdfs.client.HdfsDataInputStream; |
50 | 52 | import org.apache.hadoop.hdfs.client.HdfsDataOutputStream; |
51 | 53 | import org.apache.hadoop.hdfs.protocol.DatanodeID; |
|
55 | 57 | import org.apache.hadoop.hdfs.protocol.HdfsConstants.DatanodeReportType; |
56 | 58 | import org.apache.hadoop.hdfs.protocol.LocatedBlock; |
57 | 59 | import org.apache.hadoop.hdfs.protocol.LocatedBlocks; |
| 60 | +import org.apache.hadoop.hdfs.protocol.OpenFileEntry; |
| 61 | +import org.apache.hadoop.hdfs.protocol.OpenFilesIterator; |
58 | 62 | import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfo; |
59 | 63 | import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager; |
60 | 64 | import org.apache.hadoop.hdfs.server.blockmanagement.BlockManagerTestUtil; |
@@ -867,6 +871,69 @@ public void run() { |
867 | 871 | closedFileSet, openFilesMap, 0); |
868 | 872 | } |
869 | 873 |
|
| 874 | + /** |
| 875 | + * Verify Decommission In Progress with List Open Files |
| 876 | + * 1. start decommissioning a node (set LeavingServiceStatus) |
| 877 | + * 2. close file with decommissioning |
| 878 | + * @throws Exception |
| 879 | + */ |
| 880 | + @Test(timeout=180000) |
| 881 | + public void testDecommissionWithCloseFileAndListOpenFiles() |
| 882 | + throws Exception { |
| 883 | + LOG.info("Starting test testDecommissionWithCloseFileAndListOpenFiles"); |
| 884 | + |
| 885 | + // Disable redundancy monitor check so that open files blocking |
| 886 | + // decommission can be listed and verified. |
| 887 | + getConf().setInt( |
| 888 | + DFSConfigKeys.DFS_NAMENODE_REDUNDANCY_INTERVAL_SECONDS_KEY, 1000); |
| 889 | + getConf().setLong( |
| 890 | + DFSConfigKeys.DFS_NAMENODE_LIST_OPENFILES_NUM_RESPONSES, 1); |
| 891 | + |
| 892 | + startSimpleCluster(1, 3); |
| 893 | + FileSystem fileSys = getCluster().getFileSystem(0); |
| 894 | + FSNamesystem ns = getCluster().getNamesystem(0); |
| 895 | + Path file = new Path("/openFile"); |
| 896 | + FSDataOutputStream st = AdminStatesBaseTest.writeIncompleteFile(fileSys, |
| 897 | + file, (short)3, (short)(fileSize / blockSize)); |
| 898 | + for (DataNode d: getCluster().getDataNodes()) { |
| 899 | + DataNodeTestUtils.triggerBlockReport(d); |
| 900 | + } |
| 901 | + |
| 902 | + LocatedBlocks lbs = NameNodeAdapter.getBlockLocations( |
| 903 | + getCluster().getNameNode(0), file.toUri().getPath(), |
| 904 | + 0, blockSize * 10); |
| 905 | + DatanodeInfo dnToDecommission = lbs.getLastLocatedBlock().getLocations()[0]; |
| 906 | + |
| 907 | + DatanodeManager dm = ns.getBlockManager().getDatanodeManager(); |
| 908 | + dnToDecommission = dm.getDatanode(dnToDecommission.getDatanodeUuid()); |
| 909 | + initExcludeHost(dnToDecommission.getXferAddr()); |
| 910 | + refreshNodes(0); |
| 911 | + BlockManagerTestUtil.recheckDecommissionState(dm); |
| 912 | + waitNodeState(dnToDecommission, AdminStates.DECOMMISSION_INPROGRESS); |
| 913 | + Thread.sleep(3000); |
| 914 | + //Make sure DatanodeAdminMonitor(DatanodeAdminBackoffMonitor) At least twice run. |
| 915 | + |
| 916 | + BatchedEntries<OpenFileEntry> batchedListEntries = getCluster(). |
| 917 | + getNameNodeRpc(0).listOpenFiles(0, |
| 918 | + EnumSet.of(OpenFilesIterator.OpenFilesType.BLOCKING_DECOMMISSION), |
| 919 | + OpenFilesIterator.FILTER_PATH_DEFAULT); |
| 920 | + assertEquals(1, batchedListEntries.size()); |
| 921 | + st.close(); //close file |
| 922 | + |
| 923 | + try { |
| 924 | + batchedListEntries = getCluster().getNameNodeRpc().listOpenFiles(0, |
| 925 | + EnumSet.of(OpenFilesIterator.OpenFilesType.BLOCKING_DECOMMISSION), |
| 926 | + OpenFilesIterator.FILTER_PATH_DEFAULT); |
| 927 | + assertEquals(0, batchedListEntries.size()); |
| 928 | + } catch (NullPointerException e) { |
| 929 | + Assert.fail("Should not throw NPE when the file is not under " + |
| 930 | + "construction but has lease!"); |
| 931 | + } |
| 932 | + initExcludeHost(""); |
| 933 | + refreshNodes(0); |
| 934 | + fileSys.delete(file, false); |
| 935 | + } |
| 936 | + |
870 | 937 | @Test(timeout = 360000) |
871 | 938 | public void testDecommissionWithOpenFileAndBlockRecovery() |
872 | 939 | throws IOException, InterruptedException { |
|
0 commit comments