TestLeaseRecovery2.java
上传用户:quxuerui
上传日期:2018-01-08
资源大小:41811k
文件大小:6k
源码类别:

网格计算

开发平台:

Java

  1. /**
  2.  * Licensed to the Apache Software Foundation (ASF) under one
  3.  * or more contributor license agreements.  See the NOTICE file
  4.  * distributed with this work for additional information
  5.  * regarding copyright ownership.  The ASF licenses this file
  6.  * to you under the Apache License, Version 2.0 (the
  7.  * "License"); you may not use this file except in compliance
  8.  * with the License.  You may obtain a copy of the License at
  9.  *
  10.  *     http://www.apache.org/licenses/LICENSE-2.0
  11.  *
  12.  * Unless required by applicable law or agreed to in writing, software
  13.  * distributed under the License is distributed on an "AS IS" BASIS,
  14.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15.  * See the License for the specific language governing permissions and
  16.  * limitations under the License.
  17.  */
  18. package org.apache.hadoop.hdfs;
  19. import java.io.IOException;
  20. import org.apache.commons.logging.impl.Log4JLogger;
  21. import org.apache.hadoop.conf.Configuration;
  22. import org.apache.hadoop.fs.FSDataInputStream;
  23. import org.apache.hadoop.fs.FSDataOutputStream;
  24. import org.apache.hadoop.fs.FileSystem;
  25. import org.apache.hadoop.fs.Path;
  26. import org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException;
  27. import org.apache.hadoop.hdfs.server.datanode.DataNode;
  28. import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
  29. import org.apache.hadoop.hdfs.server.namenode.LeaseManager;
  30. import org.apache.hadoop.security.UnixUserGroupInformation;
  31. import org.apache.hadoop.security.UserGroupInformation;
  32. import org.apache.log4j.Level;
  33. public class TestLeaseRecovery2 extends junit.framework.TestCase {
  34.   {
  35.     ((Log4JLogger)DataNode.LOG).getLogger().setLevel(Level.ALL);
  36.     ((Log4JLogger)LeaseManager.LOG).getLogger().setLevel(Level.ALL);
  37.     ((Log4JLogger)FSNamesystem.LOG).getLogger().setLevel(Level.ALL);
  38.   }
  39.   static final long BLOCK_SIZE = 1024;
  40.   static final int FILE_SIZE = 1024*16;
  41.   static final short REPLICATION_NUM = (short)3;
  42.   static byte[] buffer = new byte[FILE_SIZE];
  43.   public void testBlockSynchronization() throws Exception {
  44.     final long softLease = 1000;
  45.     final long hardLease = 60 * 60 *1000;
  46.     final short repl = 3;
  47.     final Configuration conf = new Configuration();
  48.     final int bufferSize = conf.getInt("io.file.buffer.size", 4096);
  49.     conf.setLong("dfs.block.size", BLOCK_SIZE);
  50.     conf.setInt("dfs.heartbeat.interval", 1);
  51.   //  conf.setInt("io.bytes.per.checksum", 16);
  52.     MiniDFSCluster cluster = null;
  53.     byte[] actual = new byte[FILE_SIZE];
  54.     try {
  55.       cluster = new MiniDFSCluster(conf, 5, true, null);
  56.       cluster.waitActive();
  57.       //create a file
  58.       DistributedFileSystem dfs = (DistributedFileSystem)cluster.getFileSystem();
  59.       // create a random file name
  60.       String filestr = "/foo" + AppendTestUtil.nextInt();
  61.       System.out.println("filestr=" + filestr);
  62.       Path filepath = new Path(filestr);
  63.       FSDataOutputStream stm = dfs.create(filepath, true,
  64.           bufferSize, repl, BLOCK_SIZE);
  65.       assertTrue(dfs.dfs.exists(filestr));
  66.       // write random number of bytes into it.
  67.       int size = AppendTestUtil.nextInt(FILE_SIZE);
  68.       System.out.println("size=" + size);
  69.       stm.write(buffer, 0, size);
  70.       // sync file
  71.       AppendTestUtil.LOG.info("sync");
  72.       stm.sync();
  73.       AppendTestUtil.LOG.info("leasechecker.interruptAndJoin()");
  74.       dfs.dfs.leasechecker.interruptAndJoin();
  75.       // set the soft limit to be 1 second so that the
  76.       // namenode triggers lease recovery on next attempt to write-for-open.
  77.       cluster.setLeasePeriod(softLease, hardLease);
  78.       // try to re-open the file before closing the previous handle. This
  79.       // should fail but will trigger lease recovery.
  80.       {
  81.         Configuration conf2 = new Configuration(conf);
  82.         String username = UserGroupInformation.getCurrentUGI().getUserName()+"_1";
  83.         UnixUserGroupInformation.saveToConf(conf2,
  84.             UnixUserGroupInformation.UGI_PROPERTY_NAME,
  85.             new UnixUserGroupInformation(username, new String[]{"supergroup"}));
  86.         FileSystem dfs2 = FileSystem.get(conf2);
  87.   
  88.         boolean done = false;
  89.         for(int i = 0; i < 10 && !done; i++) {
  90.           AppendTestUtil.LOG.info("i=" + i);
  91.           try {
  92.             dfs2.create(filepath, false, bufferSize, repl, BLOCK_SIZE);
  93.             fail("Creation of an existing file should never succeed.");
  94.           } catch (IOException ioe) {
  95.             final String message = ioe.getMessage();
  96.             if (message.contains("file exists")) {
  97.               AppendTestUtil.LOG.info("done", ioe);
  98.               done = true;
  99.             }
  100.             else if (message.contains(AlreadyBeingCreatedException.class.getSimpleName())) {
  101.               AppendTestUtil.LOG.info("GOOD! got " + message);
  102.             }
  103.             else {
  104.               AppendTestUtil.LOG.warn("UNEXPECTED IOException", ioe);
  105.             }
  106.           }
  107.           if (!done) {
  108.             AppendTestUtil.LOG.info("sleep " + 5000 + "ms");
  109.             try {Thread.sleep(5000);} catch (InterruptedException e) {}
  110.           }
  111.         }
  112.         assertTrue(done);
  113.       }
  114.       AppendTestUtil.LOG.info("Lease for file " +  filepath + " is recovered. "
  115.           + "Validating its contents now...");
  116.       // verify that file-size matches
  117.       assertTrue("File should be " + size + " bytes, but is actually " +
  118.                  " found to be " + dfs.getFileStatus(filepath).getLen() +
  119.                  " bytes",
  120.                  dfs.getFileStatus(filepath).getLen() == size);
  121.       // verify that there is enough data to read.
  122.       System.out.println("File size is good. Now validating sizes from datanodes...");
  123.       FSDataInputStream stmin = dfs.open(filepath);
  124.       stmin.readFully(0, actual, 0, size);
  125.       stmin.close();
  126.     }
  127.     finally {
  128.       try {
  129.         if (cluster != null) {cluster.shutdown();}
  130.       } catch (Exception e) {
  131.         // ignore
  132.       }
  133.     }
  134.   }
  135. }