001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.util;
019
020import static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertFalse;
022import static org.junit.Assert.assertNotEquals;
023import static org.junit.Assert.assertNotNull;
024import static org.junit.Assert.assertNull;
025import static org.junit.Assert.assertTrue;
026
027import java.io.File;
028import java.io.IOException;
029import java.util.Random;
030import java.util.UUID;
031import org.apache.hadoop.conf.Configuration;
032import org.apache.hadoop.fs.FSDataInputStream;
033import org.apache.hadoop.fs.FSDataOutputStream;
034import org.apache.hadoop.fs.FileStatus;
035import org.apache.hadoop.fs.FileSystem;
036import org.apache.hadoop.fs.Path;
037import org.apache.hadoop.fs.permission.FsPermission;
038import org.apache.hadoop.hbase.HBaseClassTestRule;
039import org.apache.hadoop.hbase.HBaseTestingUtility;
040import org.apache.hadoop.hbase.HConstants;
041import org.apache.hadoop.hbase.HDFSBlocksDistribution;
042import org.apache.hadoop.hbase.exceptions.DeserializationException;
043import org.apache.hadoop.hbase.fs.HFileSystem;
044import org.apache.hadoop.hbase.testclassification.MediumTests;
045import org.apache.hadoop.hbase.testclassification.MiscTests;
046import org.apache.hadoop.hdfs.DFSConfigKeys;
047import org.apache.hadoop.hdfs.DFSHedgedReadMetrics;
048import org.apache.hadoop.hdfs.DFSTestUtil;
049import org.apache.hadoop.hdfs.DistributedFileSystem;
050import org.apache.hadoop.hdfs.MiniDFSCluster;
051import org.junit.Assert;
052import org.junit.Before;
053import org.junit.ClassRule;
054import org.junit.Test;
055import org.junit.experimental.categories.Category;
056import org.slf4j.Logger;
057import org.slf4j.LoggerFactory;
058
059/**
060 * Test {@link FSUtils}.
061 */
062@Category({MiscTests.class, MediumTests.class})
063public class TestFSUtils {
064
065  @ClassRule
066  public static final HBaseClassTestRule CLASS_RULE =
067      HBaseClassTestRule.forClass(TestFSUtils.class);
068
069  private static final Logger LOG = LoggerFactory.getLogger(TestFSUtils.class);
070
071  private HBaseTestingUtility htu;
072  private FileSystem fs;
073  private Configuration conf;
074
075  @Before
076  public void setUp() throws IOException {
077    htu = new HBaseTestingUtility();
078    fs = htu.getTestFileSystem();
079    conf = htu.getConfiguration();
080  }
081
082  @Test public void testIsHDFS() throws Exception {
083    assertFalse(FSUtils.isHDFS(conf));
084    MiniDFSCluster cluster = null;
085    try {
086      cluster = htu.startMiniDFSCluster(1);
087      assertTrue(FSUtils.isHDFS(conf));
088    } finally {
089      if (cluster != null) cluster.shutdown();
090    }
091  }
092
093  private void WriteDataToHDFS(FileSystem fs, Path file, int dataSize)
094    throws Exception {
095    FSDataOutputStream out = fs.create(file);
096    byte [] data = new byte[dataSize];
097    out.write(data, 0, dataSize);
098    out.close();
099  }
100
101  @Test public void testcomputeHDFSBlocksDistribution() throws Exception {
102    final int DEFAULT_BLOCK_SIZE = 1024;
103    conf.setLong("dfs.blocksize", DEFAULT_BLOCK_SIZE);
104    MiniDFSCluster cluster = null;
105    Path testFile = null;
106
107    try {
108      // set up a cluster with 3 nodes
109      String hosts[] = new String[] { "host1", "host2", "host3" };
110      cluster = htu.startMiniDFSCluster(hosts);
111      cluster.waitActive();
112      FileSystem fs = cluster.getFileSystem();
113
114      // create a file with two blocks
115      testFile = new Path("/test1.txt");
116      WriteDataToHDFS(fs, testFile, 2*DEFAULT_BLOCK_SIZE);
117
118      // given the default replication factor is 3, the same as the number of
119      // datanodes; the locality index for each host should be 100%,
120      // or getWeight for each host should be the same as getUniqueBlocksWeights
121      final long maxTime = System.currentTimeMillis() + 2000;
122      boolean ok;
123      do {
124        ok = true;
125        FileStatus status = fs.getFileStatus(testFile);
126        HDFSBlocksDistribution blocksDistribution =
127          FSUtils.computeHDFSBlocksDistribution(fs, status, 0, status.getLen());
128        long uniqueBlocksTotalWeight =
129          blocksDistribution.getUniqueBlocksTotalWeight();
130        for (String host : hosts) {
131          long weight = blocksDistribution.getWeight(host);
132          ok = (ok && uniqueBlocksTotalWeight == weight);
133        }
134      } while (!ok && System.currentTimeMillis() < maxTime);
135      assertTrue(ok);
136      } finally {
137      htu.shutdownMiniDFSCluster();
138    }
139
140
141    try {
142      // set up a cluster with 4 nodes
143      String hosts[] = new String[] { "host1", "host2", "host3", "host4" };
144      cluster = htu.startMiniDFSCluster(hosts);
145      cluster.waitActive();
146      FileSystem fs = cluster.getFileSystem();
147
148      // create a file with three blocks
149      testFile = new Path("/test2.txt");
150      WriteDataToHDFS(fs, testFile, 3*DEFAULT_BLOCK_SIZE);
151
152      // given the default replication factor is 3, we will have total of 9
153      // replica of blocks; thus the host with the highest weight should have
154      // weight == 3 * DEFAULT_BLOCK_SIZE
155      final long maxTime = System.currentTimeMillis() + 2000;
156      long weight;
157      long uniqueBlocksTotalWeight;
158      do {
159        FileStatus status = fs.getFileStatus(testFile);
160        HDFSBlocksDistribution blocksDistribution =
161          FSUtils.computeHDFSBlocksDistribution(fs, status, 0, status.getLen());
162        uniqueBlocksTotalWeight = blocksDistribution.getUniqueBlocksTotalWeight();
163
164        String tophost = blocksDistribution.getTopHosts().get(0);
165        weight = blocksDistribution.getWeight(tophost);
166
167        // NameNode is informed asynchronously, so we may have a delay. See HBASE-6175
168      } while (uniqueBlocksTotalWeight != weight && System.currentTimeMillis() < maxTime);
169      assertTrue(uniqueBlocksTotalWeight == weight);
170
171    } finally {
172      htu.shutdownMiniDFSCluster();
173    }
174
175
176    try {
177      // set up a cluster with 4 nodes
178      String hosts[] = new String[] { "host1", "host2", "host3", "host4" };
179      cluster = htu.startMiniDFSCluster(hosts);
180      cluster.waitActive();
181      FileSystem fs = cluster.getFileSystem();
182
183      // create a file with one block
184      testFile = new Path("/test3.txt");
185      WriteDataToHDFS(fs, testFile, DEFAULT_BLOCK_SIZE);
186
187      // given the default replication factor is 3, we will have total of 3
188      // replica of blocks; thus there is one host without weight
189      final long maxTime = System.currentTimeMillis() + 2000;
190      HDFSBlocksDistribution blocksDistribution;
191      do {
192        FileStatus status = fs.getFileStatus(testFile);
193        blocksDistribution = FSUtils.computeHDFSBlocksDistribution(fs, status, 0, status.getLen());
194        // NameNode is informed asynchronously, so we may have a delay. See HBASE-6175
195      }
196      while (blocksDistribution.getTopHosts().size() != 3 && System.currentTimeMillis() < maxTime);
197      assertEquals("Wrong number of hosts distributing blocks.", 3,
198        blocksDistribution.getTopHosts().size());
199    } finally {
200      htu.shutdownMiniDFSCluster();
201    }
202  }
203
204  @Test
205  public void testVersion() throws DeserializationException, IOException {
206    final Path rootdir = htu.getDataTestDir();
207    final FileSystem fs = rootdir.getFileSystem(conf);
208    assertNull(FSUtils.getVersion(fs, rootdir));
209    // Write out old format version file.  See if we can read it in and convert.
210    Path versionFile = new Path(rootdir, HConstants.VERSION_FILE_NAME);
211    FSDataOutputStream s = fs.create(versionFile);
212    final String version = HConstants.FILE_SYSTEM_VERSION;
213    s.writeUTF(version);
214    s.close();
215    assertTrue(fs.exists(versionFile));
216    FileStatus [] status = fs.listStatus(versionFile);
217    assertNotNull(status);
218    assertTrue(status.length > 0);
219    String newVersion = FSUtils.getVersion(fs, rootdir);
220    assertEquals(version.length(), newVersion.length());
221    assertEquals(version, newVersion);
222    // File will have been converted. Exercise the pb format
223    assertEquals(version, FSUtils.getVersion(fs, rootdir));
224    FSUtils.checkVersion(fs, rootdir, true);
225  }
226
227  @Test
228  public void testPermMask() throws Exception {
229    final Path rootdir = htu.getDataTestDir();
230    final FileSystem fs = rootdir.getFileSystem(conf);
231    // default fs permission
232    FsPermission defaultFsPerm = FSUtils.getFilePermissions(fs, conf,
233        HConstants.DATA_FILE_UMASK_KEY);
234    // 'hbase.data.umask.enable' is false. We will get default fs permission.
235    assertEquals(FsPermission.getFileDefault(), defaultFsPerm);
236
237    conf.setBoolean(HConstants.ENABLE_DATA_FILE_UMASK, true);
238    // first check that we don't crash if we don't have perms set
239    FsPermission defaultStartPerm = FSUtils.getFilePermissions(fs, conf,
240        HConstants.DATA_FILE_UMASK_KEY);
241    // default 'hbase.data.umask'is 000, and this umask will be used when
242    // 'hbase.data.umask.enable' is true.
243    // Therefore we will not get the real fs default in this case.
244    // Instead we will get the starting point FULL_RWX_PERMISSIONS
245    assertEquals(new FsPermission(FSUtils.FULL_RWX_PERMISSIONS), defaultStartPerm);
246
247    conf.setStrings(HConstants.DATA_FILE_UMASK_KEY, "077");
248    // now check that we get the right perms
249    FsPermission filePerm = FSUtils.getFilePermissions(fs, conf,
250        HConstants.DATA_FILE_UMASK_KEY);
251    assertEquals(new FsPermission("700"), filePerm);
252
253    // then that the correct file is created
254    Path p = new Path("target" + File.separator + UUID.randomUUID().toString());
255    try {
256      FSDataOutputStream out = FSUtils.create(conf, fs, p, filePerm, null);
257      out.close();
258      FileStatus stat = fs.getFileStatus(p);
259      assertEquals(new FsPermission("700"), stat.getPermission());
260      // and then cleanup
261    } finally {
262      fs.delete(p, true);
263    }
264  }
265
266  @Test
267  public void testDeleteAndExists() throws Exception {
268    final Path rootdir = htu.getDataTestDir();
269    final FileSystem fs = rootdir.getFileSystem(conf);
270    conf.setBoolean(HConstants.ENABLE_DATA_FILE_UMASK, true);
271    FsPermission perms = FSUtils.getFilePermissions(fs, conf, HConstants.DATA_FILE_UMASK_KEY);
272    // then that the correct file is created
273    String file = UUID.randomUUID().toString();
274    Path p = new Path(htu.getDataTestDir(), "temptarget" + File.separator + file);
275    Path p1 = new Path(htu.getDataTestDir(), "temppath" + File.separator + file);
276    try {
277      FSDataOutputStream out = FSUtils.create(conf, fs, p, perms, null);
278      out.close();
279      assertTrue("The created file should be present", FSUtils.isExists(fs, p));
280      // delete the file with recursion as false. Only the file will be deleted.
281      FSUtils.delete(fs, p, false);
282      // Create another file
283      FSDataOutputStream out1 = FSUtils.create(conf, fs, p1, perms, null);
284      out1.close();
285      // delete the file with recursion as false. Still the file only will be deleted
286      FSUtils.delete(fs, p1, true);
287      assertFalse("The created file should be present", FSUtils.isExists(fs, p1));
288      // and then cleanup
289    } finally {
290      FSUtils.delete(fs, p, true);
291      FSUtils.delete(fs, p1, true);
292    }
293  }
294
295  @Test
296  public void testFilteredStatusDoesNotThrowOnNotFound() throws Exception {
297    MiniDFSCluster cluster = htu.startMiniDFSCluster(1);
298    try {
299      assertNull(FSUtils.listStatusWithStatusFilter(cluster.getFileSystem(), new Path("definitely/doesn't/exist"), null));
300    } finally {
301      cluster.shutdown();
302    }
303
304  }
305
306  @Test
307  public void testRenameAndSetModifyTime() throws Exception {
308    MiniDFSCluster cluster = htu.startMiniDFSCluster(1);
309    assertTrue(FSUtils.isHDFS(conf));
310
311    FileSystem fs = FileSystem.get(conf);
312    Path testDir = htu.getDataTestDirOnTestFS("testArchiveFile");
313
314    String file = UUID.randomUUID().toString();
315    Path p = new Path(testDir, file);
316
317    FSDataOutputStream out = fs.create(p);
318    out.close();
319    assertTrue("The created file should be present", FSUtils.isExists(fs, p));
320
321    long expect = System.currentTimeMillis() + 1000;
322    assertNotEquals(expect, fs.getFileStatus(p).getModificationTime());
323
324    ManualEnvironmentEdge mockEnv = new ManualEnvironmentEdge();
325    mockEnv.setValue(expect);
326    EnvironmentEdgeManager.injectEdge(mockEnv);
327    try {
328      String dstFile = UUID.randomUUID().toString();
329      Path dst = new Path(testDir , dstFile);
330
331      assertTrue(FSUtils.renameAndSetModifyTime(fs, p, dst));
332      assertFalse("The moved file should not be present", FSUtils.isExists(fs, p));
333      assertTrue("The dst file should be present", FSUtils.isExists(fs, dst));
334
335      assertEquals(expect, fs.getFileStatus(dst).getModificationTime());
336      cluster.shutdown();
337    } finally {
338      EnvironmentEdgeManager.reset();
339    }
340  }
341
342  @Test
343  public void testSetStoragePolicyDefault() throws Exception {
344    verifyNoHDFSApiInvocationForDefaultPolicy();
345    verifyFileInDirWithStoragePolicy(HConstants.DEFAULT_WAL_STORAGE_POLICY);
346  }
347
348  /**
349   * Note: currently the default policy is set to defer to HDFS and this case is to verify the
350   * logic, will need to remove the check if the default policy is changed
351   */
352  private void verifyNoHDFSApiInvocationForDefaultPolicy() {
353    FileSystem testFs = new AlwaysFailSetStoragePolicyFileSystem();
354    // There should be no exception thrown when setting to default storage policy, which indicates
355    // the HDFS API hasn't been called
356    try {
357      FSUtils.setStoragePolicy(testFs, new Path("non-exist"), HConstants.DEFAULT_WAL_STORAGE_POLICY,
358        true);
359    } catch (IOException e) {
360      Assert.fail("Should have bypassed the FS API when setting default storage policy");
361    }
362    // There should be exception thrown when given non-default storage policy, which indicates the
363    // HDFS API has been called
364    try {
365      FSUtils.setStoragePolicy(testFs, new Path("non-exist"), "HOT", true);
366      Assert.fail("Should have invoked the FS API but haven't");
367    } catch (IOException e) {
368      // expected given an invalid path
369    }
370  }
371
372  class AlwaysFailSetStoragePolicyFileSystem extends DistributedFileSystem {
373    @Override
374    public void setStoragePolicy(final Path src, final String policyName)
375            throws IOException {
376      throw new IOException("The setStoragePolicy method is invoked");
377    }
378  }
379
380  /* might log a warning, but still work. (always warning on Hadoop < 2.6.0) */
381  @Test
382  public void testSetStoragePolicyValidButMaybeNotPresent() throws Exception {
383    verifyFileInDirWithStoragePolicy("ALL_SSD");
384  }
385
386  final String INVALID_STORAGE_POLICY = "1772";
387
388  /* should log a warning, but still work. (different warning on Hadoop < 2.6.0) */
389  @Test
390  public void testSetStoragePolicyInvalid() throws Exception {
391    verifyFileInDirWithStoragePolicy(INVALID_STORAGE_POLICY);
392  }
393
394  // Here instead of TestCommonFSUtils because we need a minicluster
395  private void verifyFileInDirWithStoragePolicy(final String policy) throws Exception {
396    conf.set(HConstants.WAL_STORAGE_POLICY, policy);
397
398    MiniDFSCluster cluster = htu.startMiniDFSCluster(1);
399    try {
400      assertTrue(FSUtils.isHDFS(conf));
401
402      FileSystem fs = FileSystem.get(conf);
403      Path testDir = htu.getDataTestDirOnTestFS("testArchiveFile");
404      fs.mkdirs(testDir);
405
406      String storagePolicy =
407          conf.get(HConstants.WAL_STORAGE_POLICY, HConstants.DEFAULT_WAL_STORAGE_POLICY);
408      FSUtils.setStoragePolicy(fs, testDir, storagePolicy);
409
410      String file = UUID.randomUUID().toString();
411      Path p = new Path(testDir, file);
412      WriteDataToHDFS(fs, p, 4096);
413      HFileSystem hfs = new HFileSystem(fs);
414      String policySet = hfs.getStoragePolicyName(p);
415      LOG.debug("The storage policy of path " + p + " is " + policySet);
416      if (policy.equals(HConstants.DEFER_TO_HDFS_STORAGE_POLICY)
417              || policy.equals(INVALID_STORAGE_POLICY)) {
418        String hdfsDefaultPolicy = hfs.getStoragePolicyName(hfs.getHomeDirectory());
419        LOG.debug("The default hdfs storage policy (indicated by home path: "
420                + hfs.getHomeDirectory() + ") is " + hdfsDefaultPolicy);
421        Assert.assertEquals(hdfsDefaultPolicy, policySet);
422      } else {
423        Assert.assertEquals(policy, policySet);
424      }
425      // will assert existance before deleting.
426      cleanupFile(fs, testDir);
427    } finally {
428      cluster.shutdown();
429    }
430  }
431
432  /**
433   * Ugly test that ensures we can get at the hedged read counters in dfsclient.
434   * Does a bit of preading with hedged reads enabled using code taken from hdfs TestPread.
435   * @throws Exception
436   */
437  @Test public void testDFSHedgedReadMetrics() throws Exception {
438    // Enable hedged reads and set it so the threshold is really low.
439    // Most of this test is taken from HDFS, from TestPread.
440    conf.setInt(DFSConfigKeys.DFS_DFSCLIENT_HEDGED_READ_THREADPOOL_SIZE, 5);
441    conf.setLong(DFSConfigKeys.DFS_DFSCLIENT_HEDGED_READ_THRESHOLD_MILLIS, 0);
442    conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, 4096);
443    conf.setLong(DFSConfigKeys.DFS_CLIENT_READ_PREFETCH_SIZE_KEY, 4096);
444    // Set short retry timeouts so this test runs faster
445    conf.setInt(DFSConfigKeys.DFS_CLIENT_RETRY_WINDOW_BASE, 0);
446    conf.setBoolean("dfs.datanode.transferTo.allowed", false);
447    MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(3).build();
448    // Get the metrics.  Should be empty.
449    DFSHedgedReadMetrics metrics = FSUtils.getDFSHedgedReadMetrics(conf);
450    assertEquals(0, metrics.getHedgedReadOps());
451    FileSystem fileSys = cluster.getFileSystem();
452    try {
453      Path p = new Path("preadtest.dat");
454      // We need > 1 blocks to test out the hedged reads.
455      DFSTestUtil.createFile(fileSys, p, 12 * blockSize, 12 * blockSize,
456        blockSize, (short) 3, seed);
457      pReadFile(fileSys, p);
458      cleanupFile(fileSys, p);
459      assertTrue(metrics.getHedgedReadOps() > 0);
460    } finally {
461      fileSys.close();
462      cluster.shutdown();
463    }
464  }
465
466  // Below is taken from TestPread over in HDFS.
467  static final int blockSize = 4096;
468  static final long seed = 0xDEADBEEFL;
469
470  private void pReadFile(FileSystem fileSys, Path name) throws IOException {
471    FSDataInputStream stm = fileSys.open(name);
472    byte[] expected = new byte[12 * blockSize];
473    Random rand = new Random(seed);
474    rand.nextBytes(expected);
475    // do a sanity check. Read first 4K bytes
476    byte[] actual = new byte[4096];
477    stm.readFully(actual);
478    checkAndEraseData(actual, 0, expected, "Read Sanity Test");
479    // now do a pread for the first 8K bytes
480    actual = new byte[8192];
481    doPread(stm, 0L, actual, 0, 8192);
482    checkAndEraseData(actual, 0, expected, "Pread Test 1");
483    // Now check to see if the normal read returns 4K-8K byte range
484    actual = new byte[4096];
485    stm.readFully(actual);
486    checkAndEraseData(actual, 4096, expected, "Pread Test 2");
487    // Now see if we can cross a single block boundary successfully
488    // read 4K bytes from blockSize - 2K offset
489    stm.readFully(blockSize - 2048, actual, 0, 4096);
490    checkAndEraseData(actual, (blockSize - 2048), expected, "Pread Test 3");
491    // now see if we can cross two block boundaries successfully
492    // read blockSize + 4K bytes from blockSize - 2K offset
493    actual = new byte[blockSize + 4096];
494    stm.readFully(blockSize - 2048, actual);
495    checkAndEraseData(actual, (blockSize - 2048), expected, "Pread Test 4");
496    // now see if we can cross two block boundaries that are not cached
497    // read blockSize + 4K bytes from 10*blockSize - 2K offset
498    actual = new byte[blockSize + 4096];
499    stm.readFully(10 * blockSize - 2048, actual);
500    checkAndEraseData(actual, (10 * blockSize - 2048), expected, "Pread Test 5");
501    // now check that even after all these preads, we can still read
502    // bytes 8K-12K
503    actual = new byte[4096];
504    stm.readFully(actual);
505    checkAndEraseData(actual, 8192, expected, "Pread Test 6");
506    // done
507    stm.close();
508    // check block location caching
509    stm = fileSys.open(name);
510    stm.readFully(1, actual, 0, 4096);
511    stm.readFully(4*blockSize, actual, 0, 4096);
512    stm.readFully(7*blockSize, actual, 0, 4096);
513    actual = new byte[3*4096];
514    stm.readFully(0*blockSize, actual, 0, 3*4096);
515    checkAndEraseData(actual, 0, expected, "Pread Test 7");
516    actual = new byte[8*4096];
517    stm.readFully(3*blockSize, actual, 0, 8*4096);
518    checkAndEraseData(actual, 3*blockSize, expected, "Pread Test 8");
519    // read the tail
520    stm.readFully(11*blockSize+blockSize/2, actual, 0, blockSize/2);
521    IOException res = null;
522    try { // read beyond the end of the file
523      stm.readFully(11*blockSize+blockSize/2, actual, 0, blockSize);
524    } catch (IOException e) {
525      // should throw an exception
526      res = e;
527    }
528    assertTrue("Error reading beyond file boundary.", res != null);
529
530    stm.close();
531  }
532
533  private void checkAndEraseData(byte[] actual, int from, byte[] expected, String message) {
534    for (int idx = 0; idx < actual.length; idx++) {
535      assertEquals(message+" byte "+(from+idx)+" differs. expected "+
536                        expected[from+idx]+" actual "+actual[idx],
537                        actual[idx], expected[from+idx]);
538      actual[idx] = 0;
539    }
540  }
541
542  private void doPread(FSDataInputStream stm, long position, byte[] buffer,
543      int offset, int length) throws IOException {
544    int nread = 0;
545    // long totalRead = 0;
546    // DFSInputStream dfstm = null;
547
548    /* Disable. This counts do not add up. Some issue in original hdfs tests?
549    if (stm.getWrappedStream() instanceof DFSInputStream) {
550      dfstm = (DFSInputStream) (stm.getWrappedStream());
551      totalRead = dfstm.getReadStatistics().getTotalBytesRead();
552    } */
553
554    while (nread < length) {
555      int nbytes =
556          stm.read(position + nread, buffer, offset + nread, length - nread);
557      assertTrue("Error in pread", nbytes > 0);
558      nread += nbytes;
559    }
560
561    /* Disable. This counts do not add up. Some issue in original hdfs tests?
562    if (dfstm != null) {
563      if (isHedgedRead) {
564        assertTrue("Expected read statistic to be incremented",
565          length <= dfstm.getReadStatistics().getTotalBytesRead() - totalRead);
566      } else {
567        assertEquals("Expected read statistic to be incremented", length, dfstm
568            .getReadStatistics().getTotalBytesRead() - totalRead);
569      }
570    }*/
571  }
572
573  private void cleanupFile(FileSystem fileSys, Path name) throws IOException {
574    assertTrue(fileSys.exists(name));
575    assertTrue(fileSys.delete(name, true));
576    assertTrue(!fileSys.exists(name));
577  }
578
579
580  private static final boolean STREAM_CAPABILITIES_IS_PRESENT;
581  static {
582    boolean tmp = false;
583    try {
584      Class.forName("org.apache.hadoop.fs.StreamCapabilities");
585      tmp = true;
586      LOG.debug("Test thought StreamCapabilities class was present.");
587    } catch (ClassNotFoundException exception) {
588      LOG.debug("Test didn't think StreamCapabilities class was present.");
589    } finally {
590      STREAM_CAPABILITIES_IS_PRESENT = tmp;
591    }
592  }
593
594  // Here instead of TestCommonFSUtils because we need a minicluster
595  @Test
596  public void checkStreamCapabilitiesOnHdfsDataOutputStream() throws Exception {
597    MiniDFSCluster cluster = htu.startMiniDFSCluster(1);
598    try (FileSystem filesystem = cluster.getFileSystem()) {
599      FSDataOutputStream stream = filesystem.create(new Path("/tmp/foobar"));
600      assertTrue(FSUtils.hasCapability(stream, "hsync"));
601      assertTrue(FSUtils.hasCapability(stream, "hflush"));
602      assertNotEquals("We expect HdfsDataOutputStream to say it has a dummy capability iff the " +
603          "StreamCapabilities class is not defined.",
604          STREAM_CAPABILITIES_IS_PRESENT,
605          FSUtils.hasCapability(stream, "a capability that hopefully HDFS doesn't add."));
606    } finally {
607      cluster.shutdown();
608    }
609  }
610
611}