hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractCreateTest.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.contract;

 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.ParentNotDirectoryException;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.StreamCapabilities;

 import org.junit.Test;
 import org.junit.AssumptionViolatedException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import java.io.FileNotFoundException;
 import java.io.IOException;

 import static org.apache.hadoop.fs.contract.ContractTestUtils.assertCapabilities;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.dataset;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.getFileStatusEventually;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.skip;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.touch;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.writeDataset;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.writeTextFile;
 import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.ioStatisticsSourceToString;

 /**
  * Test creating files, overwrite options etc.
  */
 public abstract class AbstractContractCreateTest extends
     AbstractFSContractTestBase {

   private static final Logger LOG =
       LoggerFactory.getLogger(AbstractContractCreateTest.class);

   /**
    * How long to wait for a path to become visible.
    */
   public static final int CREATE_TIMEOUT = 15000;

   protected Path path(String filepath, boolean useBuilder) throws IOException {
     return super.path(filepath  + (useBuilder ? "" : "-builder"));
   }

   private void testCreateNewFile(boolean useBuilder) throws Throwable {
     describe("Foundational 'create a file' test, using builder API=" +
         useBuilder);
     Path path = path("testCreateNewFile", useBuilder);
     byte[] data = dataset(256, 'a', 'z');
     writeDataset(getFileSystem(), path, data, data.length, 1024 * 1024, false,
         useBuilder);
     ContractTestUtils.verifyFileContents(getFileSystem(), path, data);
   }

   @Test
   public void testCreateNewFile() throws Throwable {
     testCreateNewFile(true);
     testCreateNewFile(false);
   }

   private void testCreateFileOverExistingFileNoOverwrite(boolean useBuilder)
       throws Throwable {
     describe("Verify overwriting an existing file fails, using builder API=" +
         useBuilder);
     Path path = path("testCreateFileOverExistingFileNoOverwrite", useBuilder);
     byte[] data = dataset(256, 'a', 'z');
     writeDataset(getFileSystem(), path, data, data.length, 1024, false);
     byte[] data2 = dataset(10 * 1024, 'A', 'Z');
     try {
       writeDataset(getFileSystem(), path, data2, data2.length, 1024, false,
           useBuilder);
       fail("writing without overwrite unexpectedly succeeded");
     } catch (FileAlreadyExistsException expected) {
       //expected
       handleExpectedException(expected);
     } catch (IOException relaxed) {
       handleRelaxedException("Creating a file over a file with overwrite==false",
                              "FileAlreadyExistsException",
                              relaxed);
     }
   }

   @Test
   public void testCreateFileOverExistingFileNoOverwrite() throws Throwable {
     testCreateFileOverExistingFileNoOverwrite(false);
     testCreateFileOverExistingFileNoOverwrite(true);
   }

   private void testOverwriteExistingFile(boolean useBuilder) throws Throwable {
     describe("Overwrite an existing file and verify the new data is there, " +
         "use builder API=" + useBuilder);
     Path path = path("testOverwriteExistingFile", useBuilder);
     byte[] data = dataset(256, 'a', 'z');
     writeDataset(getFileSystem(), path, data, data.length, 1024, false,
         useBuilder);
     ContractTestUtils.verifyFileContents(getFileSystem(), path, data);
     byte[] data2 = dataset(10 * 1024, 'A', 'Z');
     writeDataset(getFileSystem(), path, data2, data2.length, 1024, true,
         useBuilder);
     ContractTestUtils.verifyFileContents(getFileSystem(), path, data2);
   }

   /**
    * This test catches some eventual consistency problems that blobstores exhibit,
    * as we are implicitly verifying that updates are consistent. This
    * is why different file lengths and datasets are used
    */
   @Test
   public void testOverwriteExistingFile() throws Throwable {
     testOverwriteExistingFile(false);
     testOverwriteExistingFile(true);
   }

   private void testOverwriteEmptyDirectory(boolean useBuilder)
       throws Throwable {
     describe("verify trying to create a file over an empty dir fails, " +
         "use builder API=" + useBuilder);
     Path path = path("testOverwriteEmptyDirectory");
     mkdirs(path);
     assertIsDirectory(path);
     byte[] data = dataset(256, 'a', 'z');
     try {
       writeDataset(getFileSystem(), path, data, data.length, 1024, true,
           useBuilder);
       assertIsDirectory(path);
       fail("write of file over empty dir succeeded");
     } catch (FileAlreadyExistsException expected) {
       //expected
       handleExpectedException(expected);
     } catch (IOException e) {
       handleRelaxedException("overwriting a dir with a file ",
                              "FileAlreadyExistsException",
                              e);
     }
     assertIsDirectory(path);
   }

   @Test
   public void testOverwriteEmptyDirectory() throws Throwable {
     testOverwriteEmptyDirectory(false);
     testOverwriteEmptyDirectory(true);
   }

   private void testOverwriteNonEmptyDirectory(boolean useBuilder)
       throws Throwable {
     describe("verify trying to create a file over a non-empty dir fails, " +
         "use builder API=" + useBuilder);
     Path path = path("testOverwriteNonEmptyDirectory");
     mkdirs(path);
     try {
       assertIsDirectory(path);
     } catch (AssertionError failure) {
       if (isSupported(CREATE_OVERWRITES_DIRECTORY)) {
         // file/directory hack surfaces here
         throw new AssumptionViolatedException(failure.toString(), failure);
       }
       // else: rethrow
       throw failure;
     }
     Path child = new Path(path, "child");
     writeTextFile(getFileSystem(), child, "child file", true);
     byte[] data = dataset(256, 'a', 'z');
     try {
       writeDataset(getFileSystem(), path, data, data.length, 1024,
                    true, useBuilder);
       FileStatus status = getFileSystem().getFileStatus(path);

       boolean isDir = status.isDirectory();
       if (!isDir && isSupported(CREATE_OVERWRITES_DIRECTORY)) {
         // For some file systems, downgrade to a skip so that the failure is
         // visible in test results.
         skip("This Filesystem allows a file to overwrite a directory");
       }
       fail("write of file over dir succeeded");
     } catch (FileAlreadyExistsException expected) {
       //expected
       handleExpectedException(expected);
     } catch (IOException e) {
       handleRelaxedException("overwriting a dir with a file ",
                              "FileAlreadyExistsException",
                              e);
     }
     assertIsDirectory(path);
     assertIsFile(child);
   }

   @Test
   public void testOverwriteNonEmptyDirectory() throws Throwable {
     testOverwriteNonEmptyDirectory(false);
     testOverwriteNonEmptyDirectory(true);
   }

   @Test
   public void testCreatedFileIsImmediatelyVisible() throws Throwable {
     describe("verify that a newly created file exists as soon as open returns");
     Path path = path("testCreatedFileIsImmediatelyVisible");
     try(FSDataOutputStream out = getFileSystem().create(path,
                                    false,
                                    4096,
                                    (short) 1,
                                    1024)) {
       if (!getFileSystem().exists(path)) {

         if (isSupported(CREATE_VISIBILITY_DELAYED)) {
           // For some file systems, downgrade to a skip so that the failure is
           // visible in test results.
           skip("This Filesystem delays visibility of newly created files");
         }
         assertPathExists("expected path to be visible before anything written",
                          path);
       }
     }
   }

   @Test
   public void testCreatedFileIsVisibleOnFlush() throws Throwable {
     describe("verify that a newly created file exists once a flush has taken "
         + "place");
     Path path = path("testCreatedFileIsVisibleOnFlush");
     FileSystem fs = getFileSystem();
     try(FSDataOutputStream out = fs.create(path,
           false,
           4096,
           (short) 1,
           1024)) {
       out.write('a');
       out.flush();
       if (!fs.exists(path)) {
         if (isSupported(IS_BLOBSTORE) ||
             isSupported(CREATE_VISIBILITY_DELAYED)) {
           // object store or some file systems: downgrade to a skip so that the
           // failure is visible in test results
           skip("For object store or some file systems, newly created files are"
               + " not immediately visible");
         }
         assertPathExists("expected path to be visible before file closed",
             path);
       }
     }
   }

   @Test
   public void testCreatedFileIsEventuallyVisible() throws Throwable {
     describe("verify a written to file is visible after the stream is closed");
     Path path = path("testCreatedFileIsEventuallyVisible");
     FileSystem fs = getFileSystem();
     try(FSDataOutputStream out = fs.create(path, false, 4096, (short) 1,
         1024)) {
       out.write(0x01);
       out.close();
       getFileStatusEventually(fs, path, CREATE_TIMEOUT);
     }
   }

   @Test
   public void testFileStatusBlocksizeNonEmptyFile() throws Throwable {
     describe("validate the block size of a filesystem and files within it");
     FileSystem fs = getFileSystem();

     long rootPath = fs.getDefaultBlockSize(path("/"));
     assertTrue("Root block size is invalid " + rootPath,
         rootPath > 0);

     Path path = path("testFileStatusBlocksizeNonEmptyFile");
     byte[] data = dataset(256, 'a', 'z');

     writeDataset(fs, path, data, data.length, 1024 * 1024, false);

     validateBlockSize(fs, path, 1);
   }

   @Test
   public void testFileStatusBlocksizeEmptyFile() throws Throwable {
     describe("check that an empty file may return a 0-byte blocksize");
     FileSystem fs = getFileSystem();
     Path path = path("testFileStatusBlocksizeEmptyFile");
     ContractTestUtils.touch(fs, path);
     validateBlockSize(fs, path, 0);
   }

   private void validateBlockSize(FileSystem fs, Path path, int minValue)
       throws IOException, InterruptedException {
     FileStatus status =
         getFileStatusEventually(fs, path, CREATE_TIMEOUT);
     String statusDetails = status.toString();
     assertTrue("File status block size too low:  " + statusDetails
             + " min value: " + minValue,
         status.getBlockSize() >= minValue);
     long defaultBlockSize = fs.getDefaultBlockSize(path);
     assertTrue("fs.getDefaultBlockSize(" + path + ") size " +
             defaultBlockSize + " is below the minimum of " + minValue,
         defaultBlockSize >= minValue);
   }

   @Test
   public void testCreateMakesParentDirs() throws Throwable {
     describe("check that after creating a file its parent directories exist");
     FileSystem fs = getFileSystem();
     Path grandparent = path("testCreateCreatesAndPopulatesParents");
     Path parent = new Path(grandparent, "parent");
     Path child = new Path(parent, "child");
     touch(fs, child);
     assertEquals("List status of parent should include the 1 child file",
         1, fs.listStatus(parent).length);
     assertTrue("Parent directory does not appear to be a directory",
         fs.getFileStatus(parent).isDirectory());
     assertEquals("List status of grandparent should include the 1 parent dir",
         1, fs.listStatus(grandparent).length);
     assertTrue("Grandparent directory does not appear to be a directory",
         fs.getFileStatus(grandparent).isDirectory());
   }

   @Test
   public void testCreateFileUnderFile() throws Throwable {
     describe("Verify that it is forbidden to create file/file");
     if (isSupported(CREATE_FILE_UNDER_FILE_ALLOWED)) {
       // object store or some file systems: downgrade to a skip so that the
       // failure is visible in test results
       skip("This filesystem supports creating files under files");
     }
     Path grandparent = methodPath();
     Path parent = new Path(grandparent, "parent");
     expectCreateUnderFileFails(
         "creating a file under a file",
         grandparent,
         parent);
   }

   @Test
   public void testCreateUnderFileSubdir() throws Throwable {
     describe("Verify that it is forbidden to create file/dir/file");
     if (isSupported(CREATE_FILE_UNDER_FILE_ALLOWED)) {
       // object store or some file systems: downgrade to a skip so that the
       // failure is visible in test results
       skip("This filesystem supports creating files under files");
     }
     Path grandparent = methodPath();
     Path parent = new Path(grandparent, "parent");
     Path child = new Path(parent, "child");
     expectCreateUnderFileFails(
         "creating a file under a subdirectory of a file",
         grandparent,
         child);
   }


   @Test
   public void testMkdirUnderFile() throws Throwable {
     describe("Verify that it is forbidden to create file/dir");
     Path grandparent = methodPath();
     Path parent = new Path(grandparent, "parent");
     expectMkdirsUnderFileFails("mkdirs() under a file",
         grandparent, parent);
   }

   @Test
   public void testMkdirUnderFileSubdir() throws Throwable {
     describe("Verify that it is forbidden to create file/dir/dir");
     Path grandparent = methodPath();
     Path parent = new Path(grandparent, "parent");
     Path child = new Path(parent, "child");
     expectMkdirsUnderFileFails("mkdirs() file/dir",
         grandparent, child);

     try {
       // create the child
       mkdirs(child);
     } catch (FileAlreadyExistsException | ParentNotDirectoryException ex) {
       // either of these may be raised.
       handleExpectedException(ex);
     } catch (IOException e) {
       handleRelaxedException("creating a file under a subdirectory of a file ",
           "FileAlreadyExistsException",
           e);
     }
   }

   /**
    * Expect that touch() will fail because the parent is a file.
    * @param action action for message
    * @param file filename to create
    * @param descendant path under file
    * @throws Exception failure
    */
   protected void expectCreateUnderFileFails(String action,
       Path file, Path descendant)
       throws Exception {
     createFile(file);
     try {
       // create the child
       createFile(descendant);
     } catch (FileAlreadyExistsException | ParentNotDirectoryException ex) {
       //expected
       handleExpectedException(ex);
     } catch (IOException e) {
       handleRelaxedException(action,
           "ParentNotDirectoryException",
           e);
     }
   }

   protected void expectMkdirsUnderFileFails(String action,
       Path file, Path descendant)
       throws Exception {
     createFile(file);
     try {
       // now mkdirs
       mkdirs(descendant);
     } catch (FileAlreadyExistsException | ParentNotDirectoryException ex) {
       //expected
       handleExpectedException(ex);
     } catch (IOException e) {
       handleRelaxedException(action,
           "ParentNotDirectoryException",
           e);
     }
   }

   private void createFile(Path path) throws IOException {
     byte[] data = dataset(256, 'a', 'z');
     FileSystem fs = getFileSystem();
     writeDataset(fs, path, data, data.length, 1024 * 1024,
         true);
   }

   @Test
   public void testSyncable() throws Throwable {
     describe("test declared and actual Syncable behaviors");
     FileSystem fs = getFileSystem();
     boolean supportsFlush = isSupported(SUPPORTS_HFLUSH);
     boolean supportsSync = isSupported(SUPPORTS_HSYNC);
     boolean metadataUpdatedOnHSync = isSupported(METADATA_UPDATED_ON_HSYNC);

     validateSyncableSemantics(fs,
         supportsSync,
         supportsFlush,
         metadataUpdatedOnHSync);
   }

   /**
    * Validate the semantics of syncable.
    * @param fs filesystem
    * @param supportsSync sync is present
    * @param supportsFlush flush is present.
    * @param metadataUpdatedOnHSync  Is the metadata updated after an hsync?
    * @throws IOException failure
    */
   protected void validateSyncableSemantics(final FileSystem fs,
       final boolean supportsSync,
       final boolean supportsFlush,
       final boolean metadataUpdatedOnHSync)
       throws IOException {
     Path path = methodPath();
     LOG.info("Expecting files under {} to have supportsSync={}"
             + " and supportsFlush={}; metadataUpdatedOnHSync={}",
         path, supportsSync, supportsFlush, metadataUpdatedOnHSync);

     try (FSDataOutputStream out = fs.create(path, true)) {
       LOG.info("Created output stream {}", out);

       // probe stream for support for flush/sync, whose capabilities
       // of supports/does not support must match what is expected
       String[] hflushCapabilities = {
           StreamCapabilities.HFLUSH
       };
       String[] hsyncCapabilities = {
           StreamCapabilities.HSYNC
       };
       if (supportsFlush) {
         assertCapabilities(out, hflushCapabilities, null);
       } else {
         assertCapabilities(out, null, hflushCapabilities);
       }
       if (supportsSync) {
         assertCapabilities(out, hsyncCapabilities, null);
       } else {
         assertCapabilities(out, null, hsyncCapabilities);
       }

       // write one byte, then hflush it
       out.write('a');
       try {
         out.hflush();
         if (!supportsFlush) {
           // FSDataOutputStream silently downgrades to flush() here.
           // This is not good, but if changed some applications
           // break writing to some stores.
           LOG.warn("FS doesn't support Syncable.hflush(),"
               + " but doesn't reject it either.");
         }
       } catch (UnsupportedOperationException e) {
         if (supportsFlush) {
           throw new AssertionError("hflush not supported", e);
         }
       }

       // write a second byte, then hsync it.
       out.write('b');
       try {
         out.hsync();
       } catch (UnsupportedOperationException e) {
         if (supportsSync) {
           throw new AssertionError("HSync not supported", e);
         }
       }

       if (supportsSync) {
         // if sync really worked, data MUST be visible here

         // first the metadata which MUST be present
         final FileStatus st = fs.getFileStatus(path);
         if (metadataUpdatedOnHSync) {
           // not all stores reliably update it, HDFS/webHDFS in particular
           assertEquals("Metadata not updated during write " + st,
               2, st.getLen());
         }

         // there's no way to verify durability, but we can
         // at least verify a new file input stream reads
         // the data
         try (FSDataInputStream in = fs.open(path)) {
           assertEquals('a', in.read());
           assertEquals('b', in.read());
           assertEquals(-1, in.read());
           LOG.info("Successfully read synced data on a new reader {}", in);
         }
       } else {
         // no sync. Let's do a flush and see what happens.
         out.flush();
         // Now look at the filesystem.
         try (FSDataInputStream in = fs.open(path)) {
           int c = in.read();
           if (c == -1) {
             // nothing was synced; sync and flush really aren't there.
             LOG.info("sync and flush are declared unsupported"
                 + " -flushed changes were not saved");

           } else {
             LOG.info("sync and flush are declared unsupported"
                 + " - but the stream does offer some sync/flush semantics");
           }
           // close outside a finally as we do want to see any exception raised.
           in.close();

         } catch (FileNotFoundException e) {
           // that's OK if it's an object store or some file systems that newly created files
           // are not immediately visible, but not if its a real FS
           if (!isSupported(IS_BLOBSTORE) && !isSupported(CREATE_VISIBILITY_DELAYED)) {
             throw e;
           } else {
             LOG.warn(
                 "Output file was not created; this is an object store or "
                     + "a file system with different visibility semantics");
           }
         }
       }
       // close the output stream
       out.close();

       final String stats = ioStatisticsSourceToString(out);
       if (!stats.isEmpty()) {
         LOG.info("IOStatistics {}", stats);
       }
     }
   }
 }
	/*
	* 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.contract;

	import org.apache.hadoop.fs.FSDataInputStream;
	import org.apache.hadoop.fs.FSDataOutputStream;
	import org.apache.hadoop.fs.FileAlreadyExistsException;
	import org.apache.hadoop.fs.FileStatus;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.ParentNotDirectoryException;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.fs.StreamCapabilities;

	import org.junit.Test;
	import org.junit.AssumptionViolatedException;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import java.io.FileNotFoundException;
	import java.io.IOException;

	import static org.apache.hadoop.fs.contract.ContractTestUtils.assertCapabilities;
	import static org.apache.hadoop.fs.contract.ContractTestUtils.dataset;
	import static org.apache.hadoop.fs.contract.ContractTestUtils.getFileStatusEventually;
	import static org.apache.hadoop.fs.contract.ContractTestUtils.skip;
	import static org.apache.hadoop.fs.contract.ContractTestUtils.touch;
	import static org.apache.hadoop.fs.contract.ContractTestUtils.writeDataset;
	import static org.apache.hadoop.fs.contract.ContractTestUtils.writeTextFile;
	import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.ioStatisticsSourceToString;

	/**
	* Test creating files, overwrite options etc.
	*/
	public abstract class AbstractContractCreateTest extends
	AbstractFSContractTestBase {

	private static final Logger LOG =
	LoggerFactory.getLogger(AbstractContractCreateTest.class);

	/**
	* How long to wait for a path to become visible.
	*/
	public static final int CREATE_TIMEOUT = 15000;

	protected Path path(String filepath, boolean useBuilder) throws IOException {
	return super.path(filepath + (useBuilder ? "" : "-builder"));
	}

	private void testCreateNewFile(boolean useBuilder) throws Throwable {
	describe("Foundational 'create a file' test, using builder API=" +
	useBuilder);
	Path path = path("testCreateNewFile", useBuilder);
	byte[] data = dataset(256, 'a', 'z');
	writeDataset(getFileSystem(), path, data, data.length, 1024 * 1024, false,
	useBuilder);
	ContractTestUtils.verifyFileContents(getFileSystem(), path, data);
	}

	@Test
	public void testCreateNewFile() throws Throwable {
	testCreateNewFile(true);
	testCreateNewFile(false);
	}

	private void testCreateFileOverExistingFileNoOverwrite(boolean useBuilder)
	throws Throwable {
	describe("Verify overwriting an existing file fails, using builder API=" +
	useBuilder);
	Path path = path("testCreateFileOverExistingFileNoOverwrite", useBuilder);
	byte[] data = dataset(256, 'a', 'z');
	writeDataset(getFileSystem(), path, data, data.length, 1024, false);
	byte[] data2 = dataset(10 * 1024, 'A', 'Z');
	try {
	writeDataset(getFileSystem(), path, data2, data2.length, 1024, false,
	useBuilder);
	fail("writing without overwrite unexpectedly succeeded");
	} catch (FileAlreadyExistsException expected) {
	//expected
	handleExpectedException(expected);
	} catch (IOException relaxed) {
	handleRelaxedException("Creating a file over a file with overwrite==false",
	"FileAlreadyExistsException",
	relaxed);
	}
	}

	@Test
	public void testCreateFileOverExistingFileNoOverwrite() throws Throwable {
	testCreateFileOverExistingFileNoOverwrite(false);
	testCreateFileOverExistingFileNoOverwrite(true);
	}

	private void testOverwriteExistingFile(boolean useBuilder) throws Throwable {
	describe("Overwrite an existing file and verify the new data is there, " +
	"use builder API=" + useBuilder);
	Path path = path("testOverwriteExistingFile", useBuilder);
	byte[] data = dataset(256, 'a', 'z');
	writeDataset(getFileSystem(), path, data, data.length, 1024, false,
	useBuilder);
	ContractTestUtils.verifyFileContents(getFileSystem(), path, data);
	byte[] data2 = dataset(10 * 1024, 'A', 'Z');
	writeDataset(getFileSystem(), path, data2, data2.length, 1024, true,
	useBuilder);
	ContractTestUtils.verifyFileContents(getFileSystem(), path, data2);
	}

	/**
	* This test catches some eventual consistency problems that blobstores exhibit,
	* as we are implicitly verifying that updates are consistent. This
	* is why different file lengths and datasets are used
	*/
	@Test
	public void testOverwriteExistingFile() throws Throwable {
	testOverwriteExistingFile(false);
	testOverwriteExistingFile(true);
	}

	private void testOverwriteEmptyDirectory(boolean useBuilder)
	throws Throwable {
	describe("verify trying to create a file over an empty dir fails, " +
	"use builder API=" + useBuilder);
	Path path = path("testOverwriteEmptyDirectory");
	mkdirs(path);
	assertIsDirectory(path);
	byte[] data = dataset(256, 'a', 'z');
	try {
	writeDataset(getFileSystem(), path, data, data.length, 1024, true,
	useBuilder);
	assertIsDirectory(path);
	fail("write of file over empty dir succeeded");
	} catch (FileAlreadyExistsException expected) {
	//expected
	handleExpectedException(expected);
	} catch (IOException e) {
	handleRelaxedException("overwriting a dir with a file ",
	"FileAlreadyExistsException",
	e);
	}
	assertIsDirectory(path);
	}

	@Test
	public void testOverwriteEmptyDirectory() throws Throwable {
	testOverwriteEmptyDirectory(false);
	testOverwriteEmptyDirectory(true);
	}

	private void testOverwriteNonEmptyDirectory(boolean useBuilder)
	throws Throwable {
	describe("verify trying to create a file over a non-empty dir fails, " +
	"use builder API=" + useBuilder);
	Path path = path("testOverwriteNonEmptyDirectory");
	mkdirs(path);
	try {
	assertIsDirectory(path);
	} catch (AssertionError failure) {
	if (isSupported(CREATE_OVERWRITES_DIRECTORY)) {
	// file/directory hack surfaces here
	throw new AssumptionViolatedException(failure.toString(), failure);
	}
	// else: rethrow
	throw failure;
	}
	Path child = new Path(path, "child");
	writeTextFile(getFileSystem(), child, "child file", true);
	byte[] data = dataset(256, 'a', 'z');
	try {
	writeDataset(getFileSystem(), path, data, data.length, 1024,
	true, useBuilder);
	FileStatus status = getFileSystem().getFileStatus(path);

	boolean isDir = status.isDirectory();
	if (!isDir && isSupported(CREATE_OVERWRITES_DIRECTORY)) {
	// For some file systems, downgrade to a skip so that the failure is
	// visible in test results.
	skip("This Filesystem allows a file to overwrite a directory");
	}
	fail("write of file over dir succeeded");
	} catch (FileAlreadyExistsException expected) {
	//expected
	handleExpectedException(expected);
	} catch (IOException e) {
	handleRelaxedException("overwriting a dir with a file ",
	"FileAlreadyExistsException",
	e);
	}
	assertIsDirectory(path);
	assertIsFile(child);
	}

	@Test
	public void testOverwriteNonEmptyDirectory() throws Throwable {
	testOverwriteNonEmptyDirectory(false);
	testOverwriteNonEmptyDirectory(true);
	}

	@Test
	public void testCreatedFileIsImmediatelyVisible() throws Throwable {
	describe("verify that a newly created file exists as soon as open returns");
	Path path = path("testCreatedFileIsImmediatelyVisible");
	try(FSDataOutputStream out = getFileSystem().create(path,
	false,
	4096,
	(short) 1,
	1024)) {
	if (!getFileSystem().exists(path)) {

	if (isSupported(CREATE_VISIBILITY_DELAYED)) {
	// For some file systems, downgrade to a skip so that the failure is
	// visible in test results.
	skip("This Filesystem delays visibility of newly created files");
	}
	assertPathExists("expected path to be visible before anything written",
	path);
	}
	}
	}

	@Test
	public void testCreatedFileIsVisibleOnFlush() throws Throwable {
	describe("verify that a newly created file exists once a flush has taken "
	+ "place");
	Path path = path("testCreatedFileIsVisibleOnFlush");
	FileSystem fs = getFileSystem();
	try(FSDataOutputStream out = fs.create(path,
	false,
	4096,
	(short) 1,
	1024)) {
	out.write('a');
	out.flush();
	if (!fs.exists(path)) {
	if (isSupported(IS_BLOBSTORE) \|\|
	isSupported(CREATE_VISIBILITY_DELAYED)) {
	// object store or some file systems: downgrade to a skip so that the
	// failure is visible in test results
	skip("For object store or some file systems, newly created files are"
	+ " not immediately visible");
	}
	assertPathExists("expected path to be visible before file closed",
	path);
	}
	}
	}

	@Test
	public void testCreatedFileIsEventuallyVisible() throws Throwable {
	describe("verify a written to file is visible after the stream is closed");
	Path path = path("testCreatedFileIsEventuallyVisible");
	FileSystem fs = getFileSystem();
	try(FSDataOutputStream out = fs.create(path, false, 4096, (short) 1,
	1024)) {
	out.write(0x01);
	out.close();
	getFileStatusEventually(fs, path, CREATE_TIMEOUT);
	}
	}

	@Test
	public void testFileStatusBlocksizeNonEmptyFile() throws Throwable {
	describe("validate the block size of a filesystem and files within it");
	FileSystem fs = getFileSystem();

	long rootPath = fs.getDefaultBlockSize(path("/"));
	assertTrue("Root block size is invalid " + rootPath,
	rootPath > 0);

	Path path = path("testFileStatusBlocksizeNonEmptyFile");
	byte[] data = dataset(256, 'a', 'z');

	writeDataset(fs, path, data, data.length, 1024 * 1024, false);

	validateBlockSize(fs, path, 1);
	}

	@Test
	public void testFileStatusBlocksizeEmptyFile() throws Throwable {
	describe("check that an empty file may return a 0-byte blocksize");
	FileSystem fs = getFileSystem();
	Path path = path("testFileStatusBlocksizeEmptyFile");
	ContractTestUtils.touch(fs, path);
	validateBlockSize(fs, path, 0);
	}

	private void validateBlockSize(FileSystem fs, Path path, int minValue)
	throws IOException, InterruptedException {
	FileStatus status =
	getFileStatusEventually(fs, path, CREATE_TIMEOUT);
	String statusDetails = status.toString();
	assertTrue("File status block size too low: " + statusDetails
	+ " min value: " + minValue,
	status.getBlockSize() >= minValue);
	long defaultBlockSize = fs.getDefaultBlockSize(path);
	assertTrue("fs.getDefaultBlockSize(" + path + ") size " +
	defaultBlockSize + " is below the minimum of " + minValue,
	defaultBlockSize >= minValue);
	}

	@Test
	public void testCreateMakesParentDirs() throws Throwable {
	describe("check that after creating a file its parent directories exist");
	FileSystem fs = getFileSystem();
	Path grandparent = path("testCreateCreatesAndPopulatesParents");
	Path parent = new Path(grandparent, "parent");
	Path child = new Path(parent, "child");
	touch(fs, child);
	assertEquals("List status of parent should include the 1 child file",
	1, fs.listStatus(parent).length);
	assertTrue("Parent directory does not appear to be a directory",
	fs.getFileStatus(parent).isDirectory());
	assertEquals("List status of grandparent should include the 1 parent dir",
	1, fs.listStatus(grandparent).length);
	assertTrue("Grandparent directory does not appear to be a directory",
	fs.getFileStatus(grandparent).isDirectory());
	}

	@Test
	public void testCreateFileUnderFile() throws Throwable {
	describe("Verify that it is forbidden to create file/file");
	if (isSupported(CREATE_FILE_UNDER_FILE_ALLOWED)) {
	// object store or some file systems: downgrade to a skip so that the
	// failure is visible in test results
	skip("This filesystem supports creating files under files");
	}
	Path grandparent = methodPath();
	Path parent = new Path(grandparent, "parent");
	expectCreateUnderFileFails(
	"creating a file under a file",
	grandparent,
	parent);
	}

	@Test
	public void testCreateUnderFileSubdir() throws Throwable {
	describe("Verify that it is forbidden to create file/dir/file");
	if (isSupported(CREATE_FILE_UNDER_FILE_ALLOWED)) {
	// object store or some file systems: downgrade to a skip so that the
	// failure is visible in test results
	skip("This filesystem supports creating files under files");
	}
	Path grandparent = methodPath();
	Path parent = new Path(grandparent, "parent");
	Path child = new Path(parent, "child");
	expectCreateUnderFileFails(
	"creating a file under a subdirectory of a file",
	grandparent,
	child);
	}


	@Test
	public void testMkdirUnderFile() throws Throwable {
	describe("Verify that it is forbidden to create file/dir");
	Path grandparent = methodPath();
	Path parent = new Path(grandparent, "parent");
	expectMkdirsUnderFileFails("mkdirs() under a file",
	grandparent, parent);
	}

	@Test
	public void testMkdirUnderFileSubdir() throws Throwable {
	describe("Verify that it is forbidden to create file/dir/dir");
	Path grandparent = methodPath();
	Path parent = new Path(grandparent, "parent");
	Path child = new Path(parent, "child");
	expectMkdirsUnderFileFails("mkdirs() file/dir",
	grandparent, child);

	try {
	// create the child
	mkdirs(child);
	} catch (FileAlreadyExistsException \| ParentNotDirectoryException ex) {
	// either of these may be raised.
	handleExpectedException(ex);
	} catch (IOException e) {
	handleRelaxedException("creating a file under a subdirectory of a file ",
	"FileAlreadyExistsException",
	e);
	}
	}

	/**
	* Expect that touch() will fail because the parent is a file.
	* @param action action for message
	* @param file filename to create
	* @param descendant path under file
	* @throws Exception failure
	*/
	protected void expectCreateUnderFileFails(String action,
	Path file, Path descendant)
	throws Exception {
	createFile(file);
	try {
	// create the child
	createFile(descendant);
	} catch (FileAlreadyExistsException \| ParentNotDirectoryException ex) {
	//expected
	handleExpectedException(ex);
	} catch (IOException e) {
	handleRelaxedException(action,
	"ParentNotDirectoryException",
	e);
	}
	}

	protected void expectMkdirsUnderFileFails(String action,
	Path file, Path descendant)
	throws Exception {
	createFile(file);
	try {
	// now mkdirs
	mkdirs(descendant);
	} catch (FileAlreadyExistsException \| ParentNotDirectoryException ex) {
	//expected
	handleExpectedException(ex);
	} catch (IOException e) {
	handleRelaxedException(action,
	"ParentNotDirectoryException",
	e);
	}
	}

	private void createFile(Path path) throws IOException {
	byte[] data = dataset(256, 'a', 'z');
	FileSystem fs = getFileSystem();
	writeDataset(fs, path, data, data.length, 1024 * 1024,
	true);
	}

	@Test
	public void testSyncable() throws Throwable {
	describe("test declared and actual Syncable behaviors");
	FileSystem fs = getFileSystem();
	boolean supportsFlush = isSupported(SUPPORTS_HFLUSH);
	boolean supportsSync = isSupported(SUPPORTS_HSYNC);
	boolean metadataUpdatedOnHSync = isSupported(METADATA_UPDATED_ON_HSYNC);

	validateSyncableSemantics(fs,
	supportsSync,
	supportsFlush,
	metadataUpdatedOnHSync);
	}

	/**
	* Validate the semantics of syncable.
	* @param fs filesystem
	* @param supportsSync sync is present
	* @param supportsFlush flush is present.
	* @param metadataUpdatedOnHSync Is the metadata updated after an hsync?
	* @throws IOException failure
	*/
	protected void validateSyncableSemantics(final FileSystem fs,
	final boolean supportsSync,
	final boolean supportsFlush,
	final boolean metadataUpdatedOnHSync)
	throws IOException {
	Path path = methodPath();
	LOG.info("Expecting files under {} to have supportsSync={}"
	+ " and supportsFlush={}; metadataUpdatedOnHSync={}",
	path, supportsSync, supportsFlush, metadataUpdatedOnHSync);

	try (FSDataOutputStream out = fs.create(path, true)) {
	LOG.info("Created output stream {}", out);

	// probe stream for support for flush/sync, whose capabilities
	// of supports/does not support must match what is expected
	String[] hflushCapabilities = {
	StreamCapabilities.HFLUSH
	};
	String[] hsyncCapabilities = {
	StreamCapabilities.HSYNC
	};
	if (supportsFlush) {
	assertCapabilities(out, hflushCapabilities, null);
	} else {
	assertCapabilities(out, null, hflushCapabilities);
	}
	if (supportsSync) {
	assertCapabilities(out, hsyncCapabilities, null);
	} else {
	assertCapabilities(out, null, hsyncCapabilities);
	}

	// write one byte, then hflush it
	out.write('a');
	try {
	out.hflush();
	if (!supportsFlush) {
	// FSDataOutputStream silently downgrades to flush() here.
	// This is not good, but if changed some applications
	// break writing to some stores.
	LOG.warn("FS doesn't support Syncable.hflush(),"
	+ " but doesn't reject it either.");
	}
	} catch (UnsupportedOperationException e) {
	if (supportsFlush) {
	throw new AssertionError("hflush not supported", e);
	}
	}

	// write a second byte, then hsync it.
	out.write('b');
	try {
	out.hsync();
	} catch (UnsupportedOperationException e) {
	if (supportsSync) {
	throw new AssertionError("HSync not supported", e);
	}
	}

	if (supportsSync) {
	// if sync really worked, data MUST be visible here

	// first the metadata which MUST be present
	final FileStatus st = fs.getFileStatus(path);
	if (metadataUpdatedOnHSync) {
	// not all stores reliably update it, HDFS/webHDFS in particular
	assertEquals("Metadata not updated during write " + st,
	2, st.getLen());
	}

	// there's no way to verify durability, but we can
	// at least verify a new file input stream reads
	// the data
	try (FSDataInputStream in = fs.open(path)) {
	assertEquals('a', in.read());
	assertEquals('b', in.read());
	assertEquals(-1, in.read());
	LOG.info("Successfully read synced data on a new reader {}", in);
	}
	} else {
	// no sync. Let's do a flush and see what happens.
	out.flush();
	// Now look at the filesystem.
	try (FSDataInputStream in = fs.open(path)) {
	int c = in.read();
	if (c == -1) {
	// nothing was synced; sync and flush really aren't there.
	LOG.info("sync and flush are declared unsupported"
	+ " -flushed changes were not saved");

	} else {
	LOG.info("sync and flush are declared unsupported"
	+ " - but the stream does offer some sync/flush semantics");
	}
	// close outside a finally as we do want to see any exception raised.
	in.close();

	} catch (FileNotFoundException e) {
	// that's OK if it's an object store or some file systems that newly created files
	// are not immediately visible, but not if its a real FS
	if (!isSupported(IS_BLOBSTORE) && !isSupported(CREATE_VISIBILITY_DELAYED)) {
	throw e;
	} else {
	LOG.warn(
	"Output file was not created; this is an object store or "
	+ "a file system with different visibility semantics");
	}
	}
	}
	// close the output stream
	out.close();

	final String stats = ioStatisticsSourceToString(out);
	if (!stats.isEmpty()) {
	LOG.info("IOStatistics {}", stats);
	}
	}
	}
	}