public synchronized boolean saveNamespace(long timeWindow, long txGap,
      FSNamesystem source) throws IOException {
    if (timeWindow > 0 || txGap > 0) {
      final FSImageStorageInspector inspector = storage.readAndInspectDirs(
          EnumSet.of(NameNodeFile.IMAGE, NameNodeFile.IMAGE_ROLLBACK),
          StartupOption.REGULAR);
      FSImageFile image = inspector.getLatestImages().get(0);
      File imageFile = image.getFile();

      final long checkpointTxId = image.getCheckpointTxId();
      final long checkpointAge = Time.now() - imageFile.lastModified();
      if (checkpointAge <= timeWindow * 1000 &&
          checkpointTxId >= this.getCorrectLastAppliedOrWrittenTxId() - txGap) {
        return false;
      }
    }
    saveNamespace(source, NameNodeFile.IMAGE, null);
    return true;
  }
@Test
  public void testHasNonEcBlockUsingStripedIDForLoadUCFile()
      throws IOException{
    // start a cluster
    Configuration conf = new HdfsConfiguration();
    MiniDFSCluster cluster = null;
    try {
      cluster = new MiniDFSCluster.Builder(conf).numDataNodes(9)
          .build();
      cluster.waitActive();
      DistributedFileSystem fs = cluster.getFileSystem();
      FSNamesystem fns = cluster.getNamesystem();

      String testDir = "/test_block_manager";
      String testFile = "testfile_loaducfile";
      String testFilePath = testDir + "/" + testFile;
      String clientName = "testUser_loaducfile";
      String clientMachine = "testMachine_loaducfile";
      long blkId = -1;
      long blkNumBytes = 1024;
      long timestamp = 1426222918;

      fs.mkdir(new Path(testDir), new FsPermission("755"));
      Path p = new Path(testFilePath);

      DFSTestUtil.createFile(fs, p, 0, (short) 1, 1);
      BlockInfoContiguous cBlk = new BlockInfoContiguous(
          new Block(blkId, blkNumBytes, timestamp), (short)3);
      INodeFile file = (INodeFile)fns.getFSDirectory().getINode(testFilePath);
      file.toUnderConstruction(clientName, clientMachine);
      file.addBlock(cBlk);
      fns.enterSafeMode(false);
      fns.saveNamespace(0, 0);
      cluster.restartNameNodes();
      cluster.waitActive();
      fns = cluster.getNamesystem();
      assertTrue(fns.getBlockManager().hasNonEcBlockUsingStripedID());

      cluster.shutdown();
      cluster = null;
    } finally {
      if (cluster != null) {
        cluster.shutdown();
      }
    }
  }