branch-2.0.4-alpha/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestLocalDirAllocator.java - hadoop - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.hadoop.fs;

 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.NoSuchElementException;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.util.Shell;

 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
 import org.junit.Test;

 import static org.junit.Assert.*;
 import static org.junit.Assume.*;

 /** This test LocalDirAllocator works correctly;
  * Every test case uses different buffer dirs to
  * enforce the AllocatorPerContext initialization.
  * This test does not run on Cygwin because under Cygwin
  * a directory can be created in a read-only directory
  * which breaks this test.
  */
 @RunWith(Parameterized.class)
 public class TestLocalDirAllocator {
   final static private Configuration conf = new Configuration();
   final static private String BUFFER_DIR_ROOT = "build/test/temp";
   final static private String ABSOLUTE_DIR_ROOT;
   final static private String QUALIFIED_DIR_ROOT;
   final static private Path BUFFER_PATH_ROOT = new Path(BUFFER_DIR_ROOT);
   final static private File BUFFER_ROOT = new File(BUFFER_DIR_ROOT);
   final static private String CONTEXT = "mapred.local.dir";
   final static private String FILENAME = "block";
   final static private LocalDirAllocator dirAllocator =
     new LocalDirAllocator(CONTEXT);
   static LocalFileSystem localFs;
   final static private boolean isWindows =
     System.getProperty("os.name").startsWith("Windows");
   final static int SMALL_FILE_SIZE = 100;
   final static private String RELATIVE = "/RELATIVE";
   final static private String ABSOLUTE = "/ABSOLUTE";
   final static private String QUALIFIED = "/QUALIFIED";
   final private String ROOT;
   final private String PREFIX;

   static {
     try {
       localFs = FileSystem.getLocal(conf);
       rmBufferDirs();
     } catch(IOException e) {
       System.out.println(e.getMessage());
       e.printStackTrace();
       System.exit(-1);
     }

     // absolute path in test environment
     // /home/testuser/src/hadoop-common-project/hadoop-common/build/test/temp
     ABSOLUTE_DIR_ROOT = new Path(localFs.getWorkingDirectory(),
         BUFFER_DIR_ROOT).toUri().getPath();
     // file:/home/testuser/src/hadoop-common-project/hadoop-common/build/test/temp
     QUALIFIED_DIR_ROOT = new Path(localFs.getWorkingDirectory(),
         BUFFER_DIR_ROOT).toUri().toString();
   }

   public TestLocalDirAllocator(String root, String prefix) {
     ROOT = root;
     PREFIX = prefix;
   }

   @Parameters
   public static Collection<Object[]> params() {
     Object [][] data = new Object[][] {
       { BUFFER_DIR_ROOT, RELATIVE },
       { ABSOLUTE_DIR_ROOT, ABSOLUTE },
       { QUALIFIED_DIR_ROOT, QUALIFIED }
     };

     return Arrays.asList(data);
   }

   private static void rmBufferDirs() throws IOException {
     assertTrue(!localFs.exists(BUFFER_PATH_ROOT) ||
         localFs.delete(BUFFER_PATH_ROOT, true));
   }

   private static void validateTempDirCreation(String dir) throws IOException {
     File result = createTempFile(SMALL_FILE_SIZE);
     assertTrue("Checking for " + dir + " in " + result + " - FAILED!",
         result.getPath().startsWith(new Path(dir, FILENAME).toUri().getPath()));
   }

   private static File createTempFile() throws IOException {
     return createTempFile(-1);
   }

   private static File createTempFile(long size) throws IOException {
     File result = dirAllocator.createTmpFileForWrite(FILENAME, size, conf);
     result.delete();
     return result;
   }

   private String buildBufferDir(String dir, int i) {
     return dir + PREFIX + i;
   }

   /** Two buffer dirs. The first dir does not exist & is on a read-only disk;
    * The second dir exists & is RW
    * @throws Exception
    */
   @Test
   public void test0() throws Exception {
     if (isWindows) return;
     String dir0 = buildBufferDir(ROOT, 0);
     String dir1 = buildBufferDir(ROOT, 1);
     try {
       conf.set(CONTEXT, dir0 + "," + dir1);
       assertTrue(localFs.mkdirs(new Path(dir1)));
       BUFFER_ROOT.setReadOnly();
       validateTempDirCreation(dir1);
       validateTempDirCreation(dir1);
     } finally {
       Shell.execCommand(new String[]{"chmod", "u+w", BUFFER_DIR_ROOT});
       rmBufferDirs();
     }
   }

   /** Two buffer dirs. The first dir exists & is on a read-only disk;
    * The second dir exists & is RW
    * @throws Exception
    */
   @Test
   public void testROBufferDirAndRWBufferDir() throws Exception {
     if (isWindows) return;
     String dir1 = buildBufferDir(ROOT, 1);
     String dir2 = buildBufferDir(ROOT, 2);
     try {
       conf.set(CONTEXT, dir1 + "," + dir2);
       assertTrue(localFs.mkdirs(new Path(dir2)));
       BUFFER_ROOT.setReadOnly();
       validateTempDirCreation(dir2);
       validateTempDirCreation(dir2);
     } finally {
       Shell.execCommand(new String[]{"chmod", "u+w", BUFFER_DIR_ROOT});
       rmBufferDirs();
     }
   }
   /** Two buffer dirs. Both do not exist but on a RW disk.
    * Check if tmp dirs are allocated in a round-robin
    */
   @Test
   public void testDirsNotExist() throws Exception {
     if (isWindows) return;
     String dir2 = buildBufferDir(ROOT, 2);
     String dir3 = buildBufferDir(ROOT, 3);
     try {
       conf.set(CONTEXT, dir2 + "," + dir3);

       // create the first file, and then figure the round-robin sequence
       createTempFile(SMALL_FILE_SIZE);
       int firstDirIdx = (dirAllocator.getCurrentDirectoryIndex() == 0) ? 2 : 3;
       int secondDirIdx = (firstDirIdx == 2) ? 3 : 2;

       // check if tmp dirs are allocated in a round-robin manner
       validateTempDirCreation(buildBufferDir(ROOT, firstDirIdx));
       validateTempDirCreation(buildBufferDir(ROOT, secondDirIdx));
       validateTempDirCreation(buildBufferDir(ROOT, firstDirIdx));
     } finally {
       rmBufferDirs();
     }
   }

   /** Two buffer dirs. Both exists and on a R/W disk.
    * Later disk1 becomes read-only.
    * @throws Exception
    */
   @Test
   public void testRWBufferDirBecomesRO() throws Exception {
     if (isWindows) return;
     String dir3 = buildBufferDir(ROOT, 3);
     String dir4 = buildBufferDir(ROOT, 4);
     try {
       conf.set(CONTEXT, dir3 + "," + dir4);
       assertTrue(localFs.mkdirs(new Path(dir3)));
       assertTrue(localFs.mkdirs(new Path(dir4)));

       // Create the first small file
       createTempFile(SMALL_FILE_SIZE);

       // Determine the round-robin sequence
       int nextDirIdx = (dirAllocator.getCurrentDirectoryIndex() == 0) ? 3 : 4;
       validateTempDirCreation(buildBufferDir(ROOT, nextDirIdx));

       // change buffer directory 2 to be read only
       new File(new Path(dir4).toUri().getPath()).setReadOnly();
       validateTempDirCreation(dir3);
       validateTempDirCreation(dir3);
     } finally {
       rmBufferDirs();
     }
   }

   /**
    * Two buffer dirs, on read-write disk.
    *
    * Try to create a whole bunch of files.
    *  Verify that they do indeed all get created where they should.
    *
    *  Would ideally check statistical properties of distribution, but
    *  we don't have the nerve to risk false-positives here.
    *
    * @throws Exception
    */
   static final int TRIALS = 100;
   @Test
   public void testCreateManyFiles() throws Exception {
     if (isWindows) return;
     String dir5 = buildBufferDir(ROOT, 5);
     String dir6 = buildBufferDir(ROOT, 6);
     try {

       conf.set(CONTEXT, dir5 + "," + dir6);
       assertTrue(localFs.mkdirs(new Path(dir5)));
       assertTrue(localFs.mkdirs(new Path(dir6)));

       int inDir5=0, inDir6=0;
       for(int i = 0; i < TRIALS; ++i) {
         File result = createTempFile();
         if(result.getPath().startsWith(
               new Path(dir5, FILENAME).toUri().getPath())) {
           inDir5++;
         } else if(result.getPath().startsWith(
               new Path(dir6, FILENAME).toUri().getPath())) {
           inDir6++;
         }
         result.delete();
       }

       assertTrue(inDir5 + inDir6 == TRIALS);

     } finally {
       rmBufferDirs();
     }
   }

   /** Two buffer dirs. The first dir does not exist & is on a read-only disk;
    * The second dir exists & is RW
    * getLocalPathForWrite with checkAccess set to false should create a parent
    * directory. With checkAccess true, the directory should not be created.
    * @throws Exception
    */
   @Test
   public void testLocalPathForWriteDirCreation() throws IOException {
     String dir0 = buildBufferDir(ROOT, 0);
     String dir1 = buildBufferDir(ROOT, 1);
     try {
       conf.set(CONTEXT, dir0 + "," + dir1);
       assertTrue(localFs.mkdirs(new Path(dir1)));
       BUFFER_ROOT.setReadOnly();
       Path p1 =
         dirAllocator.getLocalPathForWrite("p1/x", SMALL_FILE_SIZE, conf);
       assertTrue(localFs.getFileStatus(p1.getParent()).isDirectory());

       Path p2 =
         dirAllocator.getLocalPathForWrite("p2/x", SMALL_FILE_SIZE, conf,
             false);
       try {
         localFs.getFileStatus(p2.getParent());
       } catch (Exception e) {
         assertEquals(e.getClass(), FileNotFoundException.class);
       }
     } finally {
       Shell.execCommand(new String[] { "chmod", "u+w", BUFFER_DIR_ROOT });
       rmBufferDirs();
     }
   }

   /** Test no side effect files are left over. After creating a temp
    * temp file, remove both the temp file and its parent. Verify that
    * no files or directories are left over as can happen when File objects
    * are mistakenly created from fully qualified path strings.
    * @throws IOException
    */
   @Test
   public void testNoSideEffects() throws IOException {
     assumeTrue(!isWindows);
     String dir = buildBufferDir(ROOT, 0);
     try {
       conf.set(CONTEXT, dir);
       File result = dirAllocator.createTmpFileForWrite(FILENAME, -1, conf);
       assertTrue(result.delete());
       assertTrue(result.getParentFile().delete());
       assertFalse(new File(dir).exists());
     } finally {
       Shell.execCommand(new String[]{"chmod", "u+w", BUFFER_DIR_ROOT});
       rmBufferDirs();
     }
   }

   /**
    * Test getLocalPathToRead() returns correct filename and "file" schema.
    *
    * @throws IOException
    */
   @Test
   public void testGetLocalPathToRead() throws IOException {
     assumeTrue(!isWindows);
     String dir = buildBufferDir(ROOT, 0);
     try {
       conf.set(CONTEXT, dir);
       assertTrue(localFs.mkdirs(new Path(dir)));
       File f1 = dirAllocator.createTmpFileForWrite(FILENAME, SMALL_FILE_SIZE,
           conf);
       Path p1 = dirAllocator.getLocalPathToRead(f1.getName(), conf);
       assertEquals(f1.getName(), p1.getName());
       assertEquals("file", p1.getFileSystem(conf).getUri().getScheme());
     } finally {
       Shell.execCommand(new String[] { "chmod", "u+w", BUFFER_DIR_ROOT });
       rmBufferDirs();
     }
   }

   /**
    * Test that {@link LocalDirAllocator#getAllLocalPathsToRead(String, Configuration)}
    * returns correct filenames and "file" schema.
    *
    * @throws IOException
    */
   @Test
   public void testGetAllLocalPathsToRead() throws IOException {
     assumeTrue(!isWindows);

     String dir0 = buildBufferDir(ROOT, 0);
     String dir1 = buildBufferDir(ROOT, 1);
     try {
       conf.set(CONTEXT, dir0 + "," + dir1);
       assertTrue(localFs.mkdirs(new Path(dir0)));
       assertTrue(localFs.mkdirs(new Path(dir1)));

       localFs.create(new Path(dir0 + Path.SEPARATOR + FILENAME));
       localFs.create(new Path(dir1 + Path.SEPARATOR + FILENAME));

       // check both the paths are returned as paths to read:
       final Iterable<Path> pathIterable = dirAllocator.getAllLocalPathsToRead(FILENAME, conf);
       int count = 0;
       for (final Path p: pathIterable) {
         count++;
         assertEquals(FILENAME, p.getName());
         assertEquals("file", p.getFileSystem(conf).getUri().getScheme());
       }
       assertEquals(2, count);

       // test #next() while no element to iterate any more:
       try {
         Path p = pathIterable.iterator().next();
         assertFalse("NoSuchElementException must be thrown, but returned ["+p
             +"] instead.", true); // exception expected
       } catch (NoSuchElementException nsee) {
         // okay
       }

       // test modification not allowed:
       final Iterable<Path> pathIterable2 = dirAllocator.getAllLocalPathsToRead(FILENAME, conf);
       final Iterator<Path> it = pathIterable2.iterator();
       try {
         it.remove();
         assertFalse(true); // exception expected
       } catch (UnsupportedOperationException uoe) {
         // okay
       }
     } finally {
       Shell.execCommand(new String[] { "chmod", "u+w", BUFFER_DIR_ROOT });
       rmBufferDirs();
     }
   }

   @Test
   public void testRemoveContext() throws IOException {
     String dir = buildBufferDir(ROOT, 0);
     try {
       String contextCfgItemName = "application_1340842292563_0004.app.cache.dirs";
       conf.set(contextCfgItemName, dir);
       LocalDirAllocator localDirAllocator = new LocalDirAllocator(
           contextCfgItemName);
       localDirAllocator.getLocalPathForWrite("p1/x", SMALL_FILE_SIZE, conf);
       assertTrue(LocalDirAllocator.isContextValid(contextCfgItemName));
       LocalDirAllocator.removeContext(contextCfgItemName);
       assertFalse(LocalDirAllocator.isContextValid(contextCfgItemName));
     } finally {
       rmBufferDirs();
     }
   }

 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.hadoop.fs;

	import java.io.File;
	import java.io.FileNotFoundException;
	import java.io.IOException;
	import java.util.Arrays;
	import java.util.Collection;
	import java.util.Iterator;
	import java.util.NoSuchElementException;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.util.Shell;

	import org.junit.runner.RunWith;
	import org.junit.runners.Parameterized;
	import org.junit.runners.Parameterized.Parameters;
	import org.junit.Test;

	import static org.junit.Assert.*;
	import static org.junit.Assume.*;

	/** This test LocalDirAllocator works correctly;
	* Every test case uses different buffer dirs to
	* enforce the AllocatorPerContext initialization.
	* This test does not run on Cygwin because under Cygwin
	* a directory can be created in a read-only directory
	* which breaks this test.
	*/
	@RunWith(Parameterized.class)
	public class TestLocalDirAllocator {
	final static private Configuration conf = new Configuration();
	final static private String BUFFER_DIR_ROOT = "build/test/temp";
	final static private String ABSOLUTE_DIR_ROOT;
	final static private String QUALIFIED_DIR_ROOT;
	final static private Path BUFFER_PATH_ROOT = new Path(BUFFER_DIR_ROOT);
	final static private File BUFFER_ROOT = new File(BUFFER_DIR_ROOT);
	final static private String CONTEXT = "mapred.local.dir";
	final static private String FILENAME = "block";
	final static private LocalDirAllocator dirAllocator =
	new LocalDirAllocator(CONTEXT);
	static LocalFileSystem localFs;
	final static private boolean isWindows =
	System.getProperty("os.name").startsWith("Windows");
	final static int SMALL_FILE_SIZE = 100;
	final static private String RELATIVE = "/RELATIVE";
	final static private String ABSOLUTE = "/ABSOLUTE";
	final static private String QUALIFIED = "/QUALIFIED";
	final private String ROOT;
	final private String PREFIX;

	static {
	try {
	localFs = FileSystem.getLocal(conf);
	rmBufferDirs();
	} catch(IOException e) {
	System.out.println(e.getMessage());
	e.printStackTrace();
	System.exit(-1);
	}

	// absolute path in test environment
	// /home/testuser/src/hadoop-common-project/hadoop-common/build/test/temp
	ABSOLUTE_DIR_ROOT = new Path(localFs.getWorkingDirectory(),
	BUFFER_DIR_ROOT).toUri().getPath();
	// file:/home/testuser/src/hadoop-common-project/hadoop-common/build/test/temp
	QUALIFIED_DIR_ROOT = new Path(localFs.getWorkingDirectory(),
	BUFFER_DIR_ROOT).toUri().toString();
	}

	public TestLocalDirAllocator(String root, String prefix) {
	ROOT = root;
	PREFIX = prefix;
	}

	@Parameters
	public static Collection<Object[]> params() {
	Object [][] data = new Object[][] {
	{ BUFFER_DIR_ROOT, RELATIVE },
	{ ABSOLUTE_DIR_ROOT, ABSOLUTE },
	{ QUALIFIED_DIR_ROOT, QUALIFIED }
	};

	return Arrays.asList(data);
	}

	private static void rmBufferDirs() throws IOException {
	assertTrue(!localFs.exists(BUFFER_PATH_ROOT) \|\|
	localFs.delete(BUFFER_PATH_ROOT, true));
	}

	private static void validateTempDirCreation(String dir) throws IOException {
	File result = createTempFile(SMALL_FILE_SIZE);
	assertTrue("Checking for " + dir + " in " + result + " - FAILED!",
	result.getPath().startsWith(new Path(dir, FILENAME).toUri().getPath()));
	}

	private static File createTempFile() throws IOException {
	return createTempFile(-1);
	}

	private static File createTempFile(long size) throws IOException {
	File result = dirAllocator.createTmpFileForWrite(FILENAME, size, conf);
	result.delete();
	return result;
	}

	private String buildBufferDir(String dir, int i) {
	return dir + PREFIX + i;
	}

	/** Two buffer dirs. The first dir does not exist & is on a read-only disk;
	* The second dir exists & is RW
	* @throws Exception
	*/
	@Test
	public void test0() throws Exception {
	if (isWindows) return;
	String dir0 = buildBufferDir(ROOT, 0);
	String dir1 = buildBufferDir(ROOT, 1);
	try {
	conf.set(CONTEXT, dir0 + "," + dir1);
	assertTrue(localFs.mkdirs(new Path(dir1)));
	BUFFER_ROOT.setReadOnly();
	validateTempDirCreation(dir1);
	validateTempDirCreation(dir1);
	} finally {
	Shell.execCommand(new String[]{"chmod", "u+w", BUFFER_DIR_ROOT});
	rmBufferDirs();
	}
	}

	/** Two buffer dirs. The first dir exists & is on a read-only disk;
	* The second dir exists & is RW
	* @throws Exception
	*/
	@Test
	public void testROBufferDirAndRWBufferDir() throws Exception {
	if (isWindows) return;
	String dir1 = buildBufferDir(ROOT, 1);
	String dir2 = buildBufferDir(ROOT, 2);
	try {
	conf.set(CONTEXT, dir1 + "," + dir2);
	assertTrue(localFs.mkdirs(new Path(dir2)));
	BUFFER_ROOT.setReadOnly();
	validateTempDirCreation(dir2);
	validateTempDirCreation(dir2);
	} finally {
	Shell.execCommand(new String[]{"chmod", "u+w", BUFFER_DIR_ROOT});
	rmBufferDirs();
	}
	}
	/** Two buffer dirs. Both do not exist but on a RW disk.
	* Check if tmp dirs are allocated in a round-robin
	*/
	@Test
	public void testDirsNotExist() throws Exception {
	if (isWindows) return;
	String dir2 = buildBufferDir(ROOT, 2);
	String dir3 = buildBufferDir(ROOT, 3);
	try {
	conf.set(CONTEXT, dir2 + "," + dir3);

	// create the first file, and then figure the round-robin sequence
	createTempFile(SMALL_FILE_SIZE);
	int firstDirIdx = (dirAllocator.getCurrentDirectoryIndex() == 0) ? 2 : 3;
	int secondDirIdx = (firstDirIdx == 2) ? 3 : 2;

	// check if tmp dirs are allocated in a round-robin manner
	validateTempDirCreation(buildBufferDir(ROOT, firstDirIdx));
	validateTempDirCreation(buildBufferDir(ROOT, secondDirIdx));
	validateTempDirCreation(buildBufferDir(ROOT, firstDirIdx));
	} finally {
	rmBufferDirs();
	}
	}

	/** Two buffer dirs. Both exists and on a R/W disk.
	* Later disk1 becomes read-only.
	* @throws Exception
	*/
	@Test
	public void testRWBufferDirBecomesRO() throws Exception {
	if (isWindows) return;
	String dir3 = buildBufferDir(ROOT, 3);
	String dir4 = buildBufferDir(ROOT, 4);
	try {
	conf.set(CONTEXT, dir3 + "," + dir4);
	assertTrue(localFs.mkdirs(new Path(dir3)));
	assertTrue(localFs.mkdirs(new Path(dir4)));

	// Create the first small file
	createTempFile(SMALL_FILE_SIZE);

	// Determine the round-robin sequence
	int nextDirIdx = (dirAllocator.getCurrentDirectoryIndex() == 0) ? 3 : 4;
	validateTempDirCreation(buildBufferDir(ROOT, nextDirIdx));

	// change buffer directory 2 to be read only
	new File(new Path(dir4).toUri().getPath()).setReadOnly();
	validateTempDirCreation(dir3);
	validateTempDirCreation(dir3);
	} finally {
	rmBufferDirs();
	}
	}

	/**
	* Two buffer dirs, on read-write disk.
	*
	* Try to create a whole bunch of files.
	* Verify that they do indeed all get created where they should.
	*
	* Would ideally check statistical properties of distribution, but
	* we don't have the nerve to risk false-positives here.
	*
	* @throws Exception
	*/
	static final int TRIALS = 100;
	@Test
	public void testCreateManyFiles() throws Exception {
	if (isWindows) return;
	String dir5 = buildBufferDir(ROOT, 5);
	String dir6 = buildBufferDir(ROOT, 6);
	try {

	conf.set(CONTEXT, dir5 + "," + dir6);
	assertTrue(localFs.mkdirs(new Path(dir5)));
	assertTrue(localFs.mkdirs(new Path(dir6)));

	int inDir5=0, inDir6=0;
	for(int i = 0; i < TRIALS; ++i) {
	File result = createTempFile();
	if(result.getPath().startsWith(
	new Path(dir5, FILENAME).toUri().getPath())) {
	inDir5++;
	} else if(result.getPath().startsWith(
	new Path(dir6, FILENAME).toUri().getPath())) {
	inDir6++;
	}
	result.delete();
	}

	assertTrue(inDir5 + inDir6 == TRIALS);

	} finally {
	rmBufferDirs();
	}
	}

	/** Two buffer dirs. The first dir does not exist & is on a read-only disk;
	* The second dir exists & is RW
	* getLocalPathForWrite with checkAccess set to false should create a parent
	* directory. With checkAccess true, the directory should not be created.
	* @throws Exception
	*/
	@Test
	public void testLocalPathForWriteDirCreation() throws IOException {
	String dir0 = buildBufferDir(ROOT, 0);
	String dir1 = buildBufferDir(ROOT, 1);
	try {
	conf.set(CONTEXT, dir0 + "," + dir1);
	assertTrue(localFs.mkdirs(new Path(dir1)));
	BUFFER_ROOT.setReadOnly();
	Path p1 =
	dirAllocator.getLocalPathForWrite("p1/x", SMALL_FILE_SIZE, conf);
	assertTrue(localFs.getFileStatus(p1.getParent()).isDirectory());

	Path p2 =
	dirAllocator.getLocalPathForWrite("p2/x", SMALL_FILE_SIZE, conf,
	false);
	try {
	localFs.getFileStatus(p2.getParent());
	} catch (Exception e) {
	assertEquals(e.getClass(), FileNotFoundException.class);
	}
	} finally {
	Shell.execCommand(new String[] { "chmod", "u+w", BUFFER_DIR_ROOT });
	rmBufferDirs();
	}
	}

	/** Test no side effect files are left over. After creating a temp
	* temp file, remove both the temp file and its parent. Verify that
	* no files or directories are left over as can happen when File objects
	* are mistakenly created from fully qualified path strings.
	* @throws IOException
	*/
	@Test
	public void testNoSideEffects() throws IOException {
	assumeTrue(!isWindows);
	String dir = buildBufferDir(ROOT, 0);
	try {
	conf.set(CONTEXT, dir);
	File result = dirAllocator.createTmpFileForWrite(FILENAME, -1, conf);
	assertTrue(result.delete());
	assertTrue(result.getParentFile().delete());
	assertFalse(new File(dir).exists());
	} finally {
	Shell.execCommand(new String[]{"chmod", "u+w", BUFFER_DIR_ROOT});
	rmBufferDirs();
	}
	}

	/**
	* Test getLocalPathToRead() returns correct filename and "file" schema.
	*
	* @throws IOException
	*/
	@Test
	public void testGetLocalPathToRead() throws IOException {
	assumeTrue(!isWindows);
	String dir = buildBufferDir(ROOT, 0);
	try {
	conf.set(CONTEXT, dir);
	assertTrue(localFs.mkdirs(new Path(dir)));
	File f1 = dirAllocator.createTmpFileForWrite(FILENAME, SMALL_FILE_SIZE,
	conf);
	Path p1 = dirAllocator.getLocalPathToRead(f1.getName(), conf);
	assertEquals(f1.getName(), p1.getName());
	assertEquals("file", p1.getFileSystem(conf).getUri().getScheme());
	} finally {
	Shell.execCommand(new String[] { "chmod", "u+w", BUFFER_DIR_ROOT });
	rmBufferDirs();
	}
	}

	/**
	* Test that {@link LocalDirAllocator#getAllLocalPathsToRead(String, Configuration)}
	* returns correct filenames and "file" schema.
	*
	* @throws IOException
	*/
	@Test
	public void testGetAllLocalPathsToRead() throws IOException {
	assumeTrue(!isWindows);

	String dir0 = buildBufferDir(ROOT, 0);
	String dir1 = buildBufferDir(ROOT, 1);
	try {
	conf.set(CONTEXT, dir0 + "," + dir1);
	assertTrue(localFs.mkdirs(new Path(dir0)));
	assertTrue(localFs.mkdirs(new Path(dir1)));

	localFs.create(new Path(dir0 + Path.SEPARATOR + FILENAME));
	localFs.create(new Path(dir1 + Path.SEPARATOR + FILENAME));

	// check both the paths are returned as paths to read:
	final Iterable<Path> pathIterable = dirAllocator.getAllLocalPathsToRead(FILENAME, conf);
	int count = 0;
	for (final Path p: pathIterable) {
	count++;
	assertEquals(FILENAME, p.getName());
	assertEquals("file", p.getFileSystem(conf).getUri().getScheme());
	}
	assertEquals(2, count);

	// test #next() while no element to iterate any more:
	try {
	Path p = pathIterable.iterator().next();
	assertFalse("NoSuchElementException must be thrown, but returned ["+p
	+"] instead.", true); // exception expected
	} catch (NoSuchElementException nsee) {
	// okay
	}

	// test modification not allowed:
	final Iterable<Path> pathIterable2 = dirAllocator.getAllLocalPathsToRead(FILENAME, conf);
	final Iterator<Path> it = pathIterable2.iterator();
	try {
	it.remove();
	assertFalse(true); // exception expected
	} catch (UnsupportedOperationException uoe) {
	// okay
	}
	} finally {
	Shell.execCommand(new String[] { "chmod", "u+w", BUFFER_DIR_ROOT });
	rmBufferDirs();
	}
	}

	@Test
	public void testRemoveContext() throws IOException {
	String dir = buildBufferDir(ROOT, 0);
	try {
	String contextCfgItemName = "application_1340842292563_0004.app.cache.dirs";
	conf.set(contextCfgItemName, dir);
	LocalDirAllocator localDirAllocator = new LocalDirAllocator(
	contextCfgItemName);
	localDirAllocator.getLocalPathForWrite("p1/x", SMALL_FILE_SIZE, conf);
	assertTrue(LocalDirAllocator.isContextValid(contextCfgItemName));
	LocalDirAllocator.removeContext(contextCfgItemName);
	assertFalse(LocalDirAllocator.isContextValid(contextCfgItemName));
	} finally {
	rmBufferDirs();
	}
	}

	}