@Test(timeout=10000) public void testDFSClientPeerTimeout() throws IOException { final int timeout = 1000; final Configuration conf = new HdfsConfiguration(); conf.setInt(DFSConfigKeys.DFS_CLIENT_SOCKET_TIMEOUT_KEY, timeout); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).build(); try { cluster.waitActive(); DistributedFileSystem dfs = cluster.getFileSystem(); ServerSocket socket = new ServerSocket(0); Peer peer = dfs.getClient().newConnectedPeer( (InetSocketAddress) socket.getLocalSocketAddress(), null, null); long start = Time.now(); try { peer.getInputStream().read(); Assert.fail("should timeout"); } catch (SocketTimeoutException ste) { long delta = Time.now() - start; Assert.assertTrue("timedout too soon", delta >= timeout*0.9); Assert.assertTrue("timedout too late", delta <= timeout*1.1); } catch (Throwable t) { Assert.fail("wrong exception:"+t); } } finally { cluster.shutdown(); } }
@InterfaceAudience.Private @VisibleForTesting public DFSClient getClient() { return dfs; }
DistributedFileSystem extends FileSystem { @InterfaceAudience.Private @VisibleForTesting public DFSClient getClient() { return dfs; } }
DistributedFileSystem extends FileSystem { @InterfaceAudience.Private @VisibleForTesting public DFSClient getClient() { return dfs; } DistributedFileSystem(); }
DistributedFileSystem extends FileSystem { @InterfaceAudience.Private @VisibleForTesting public DFSClient getClient() { return dfs; } DistributedFileSystem(); @Override String getScheme(); @Override URI getUri(); @Override void initialize(URI uri, Configuration conf); @Override Path getWorkingDirectory(); @Override long getDefaultBlockSize(); @Override short getDefaultReplication(); @Override void setWorkingDirectory(Path dir); @Override Path getHomeDirectory(); @Override BlockLocation[] getFileBlockLocations(FileStatus file, long start,
      long len); @Override BlockLocation[] getFileBlockLocations(Path p, 
      final long start, final long len); @InterfaceStability.Unstable BlockStorageLocation[] getFileBlockStorageLocations(
      List<BlockLocation> blocks); @Override void setVerifyChecksum(boolean verifyChecksum); boolean recoverLease(final Path f); @Override FSDataInputStream open(Path f, final int bufferSize); @Override FSDataOutputStream append(Path f, final int bufferSize,
      final Progressable progress); @Override FSDataOutputStream create(Path f, FsPermission permission,
      boolean overwrite, int bufferSize, short replication, long blockSize,
      Progressable progress); HdfsDataOutputStream create(final Path f,
      final FsPermission permission, final boolean overwrite,
      final int bufferSize, final short replication, final long blockSize,
      final Progressable progress, final InetSocketAddress[] favoredNodes); @Override FSDataOutputStream create(final Path f, final FsPermission permission,
    final EnumSet<CreateFlag> cflags, final int bufferSize,
    final short replication, final long blockSize, final Progressable progress,
    final ChecksumOpt checksumOpt); @Override @SuppressWarnings("deprecation") FSDataOutputStream createNonRecursive(final Path f,
      final FsPermission permission, final EnumSet<CreateFlag> flag,
      final int bufferSize, final short replication, final long blockSize,
      final Progressable progress); @Override boolean setReplication(Path src, 
                                final short replication
                               ); void setStoragePolicy(final Path src, final String policyName); BlockStoragePolicy[] getStoragePolicies(); @Override void concat(Path trg, Path [] psrcs); @SuppressWarnings("deprecation") @Override boolean rename(Path src, Path dst); @SuppressWarnings("deprecation") @Override void rename(Path src, Path dst, final Options.Rename... options); @Override boolean delete(Path f, final boolean recursive); @Override ContentSummary getContentSummary(Path f); void setQuota(Path src, final long namespaceQuota,
      final long diskspaceQuota); @Override FileStatus[] listStatus(Path p); boolean mkdir(Path f, FsPermission permission); @Override boolean mkdirs(Path f, FsPermission permission); @Override void close(); @Override String toString(); @InterfaceAudience.Private @VisibleForTesting DFSClient getClient(); @Override FsStatus getStatus(Path p); @Deprecated DiskStatus getDiskStatus(); @Deprecated long getRawCapacity(); @Deprecated long getRawUsed(); long getMissingBlocksCount(); long getUnderReplicatedBlocksCount(); long getCorruptBlocksCount(); @Override RemoteIterator<Path> listCorruptFileBlocks(Path path); DatanodeInfo[] getDataNodeStats(); DatanodeInfo[] getDataNodeStats(final DatanodeReportType type
      ); boolean setSafeMode(HdfsConstants.SafeModeAction action); boolean setSafeMode(HdfsConstants.SafeModeAction action,
      boolean isChecked); void saveNamespace(); long rollEdits(); boolean restoreFailedStorage(String arg); void refreshNodes(); void finalizeUpgrade(); RollingUpgradeInfo rollingUpgrade(RollingUpgradeAction action); void metaSave(String pathname); @Override FsServerDefaults getServerDefaults(); @Override FileStatus getFileStatus(Path f); @SuppressWarnings("deprecation") @Override void createSymlink(final Path target, final Path link,
      final boolean createParent); @Override boolean supportsSymlinks(); @Override FileStatus getFileLinkStatus(final Path f); @Override Path getLinkTarget(final Path f); @Override FileChecksum getFileChecksum(Path f); @Override FileChecksum getFileChecksum(Path f, final long length); @Override void setPermission(Path p, final FsPermission permission
      ); @Override void setOwner(Path p, final String username, final String groupname
      ); @Override void setTimes(Path p, final long mtime, final long atime
      ); @Override Token<DelegationTokenIdentifier> getDelegationToken(String renewer); void setBalancerBandwidth(long bandwidth); @Override String getCanonicalServiceName(); boolean isInSafeMode(); void allowSnapshot(final Path path); void disallowSnapshot(final Path path); @Override Path createSnapshot(final Path path, final String snapshotName); @Override void renameSnapshot(final Path path, final String snapshotOldName,
      final String snapshotNewName); SnapshottableDirectoryStatus[] getSnapshottableDirListing(); @Override void deleteSnapshot(final Path snapshotDir, final String snapshotName); SnapshotDiffReport getSnapshotDiffReport(final Path snapshotDir,
      final String fromSnapshot, final String toSnapshot); boolean isFileClosed(final Path src); long addCacheDirective(CacheDirectiveInfo info); long addCacheDirective(
      CacheDirectiveInfo info, EnumSet<CacheFlag> flags); void modifyCacheDirective(CacheDirectiveInfo info); void modifyCacheDirective(
      CacheDirectiveInfo info, EnumSet<CacheFlag> flags); void removeCacheDirective(long id); RemoteIterator<CacheDirectiveEntry> listCacheDirectives(
      CacheDirectiveInfo filter); void addCachePool(CachePoolInfo info); void modifyCachePool(CachePoolInfo info); void removeCachePool(String poolName); RemoteIterator<CachePoolEntry> listCachePools(); @Override void modifyAclEntries(Path path, final List<AclEntry> aclSpec); @Override void removeAclEntries(Path path, final List<AclEntry> aclSpec); @Override void removeDefaultAcl(Path path); @Override void removeAcl(Path path); @Override void setAcl(Path path, final List<AclEntry> aclSpec); @Override AclStatus getAclStatus(Path path); void createEncryptionZone(Path path, String keyName); EncryptionZone getEZForPath(Path path); RemoteIterator<EncryptionZone> listEncryptionZones(); @Override void setXAttr(Path path, final String name, final byte[] value, 
      final EnumSet<XAttrSetFlag> flag); @Override byte[] getXAttr(Path path, final String name); @Override Map<String, byte[]> getXAttrs(Path path); @Override Map<String, byte[]> getXAttrs(Path path, final List<String> names); @Override List<String> listXAttrs(Path path); @Override void removeXAttr(Path path, final String name); @Override void access(Path path, final FsAction mode); @Override Token<?>[] addDelegationTokens(
      final String renewer, Credentials credentials); DFSInotifyEventInputStream getInotifyEventStream(); DFSInotifyEventInputStream getInotifyEventStream(long lastReadTxid); }
DistributedFileSystem extends FileSystem { @InterfaceAudience.Private @VisibleForTesting public DFSClient getClient() { return dfs; } DistributedFileSystem(); @Override String getScheme(); @Override URI getUri(); @Override void initialize(URI uri, Configuration conf); @Override Path getWorkingDirectory(); @Override long getDefaultBlockSize(); @Override short getDefaultReplication(); @Override void setWorkingDirectory(Path dir); @Override Path getHomeDirectory(); @Override BlockLocation[] getFileBlockLocations(FileStatus file, long start,
      long len); @Override BlockLocation[] getFileBlockLocations(Path p, 
      final long start, final long len); @InterfaceStability.Unstable BlockStorageLocation[] getFileBlockStorageLocations(
      List<BlockLocation> blocks); @Override void setVerifyChecksum(boolean verifyChecksum); boolean recoverLease(final Path f); @Override FSDataInputStream open(Path f, final int bufferSize); @Override FSDataOutputStream append(Path f, final int bufferSize,
      final Progressable progress); @Override FSDataOutputStream create(Path f, FsPermission permission,
      boolean overwrite, int bufferSize, short replication, long blockSize,
      Progressable progress); HdfsDataOutputStream create(final Path f,
      final FsPermission permission, final boolean overwrite,
      final int bufferSize, final short replication, final long blockSize,
      final Progressable progress, final InetSocketAddress[] favoredNodes); @Override FSDataOutputStream create(final Path f, final FsPermission permission,
    final EnumSet<CreateFlag> cflags, final int bufferSize,
    final short replication, final long blockSize, final Progressable progress,
    final ChecksumOpt checksumOpt); @Override @SuppressWarnings("deprecation") FSDataOutputStream createNonRecursive(final Path f,
      final FsPermission permission, final EnumSet<CreateFlag> flag,
      final int bufferSize, final short replication, final long blockSize,
      final Progressable progress); @Override boolean setReplication(Path src, 
                                final short replication
                               ); void setStoragePolicy(final Path src, final String policyName); BlockStoragePolicy[] getStoragePolicies(); @Override void concat(Path trg, Path [] psrcs); @SuppressWarnings("deprecation") @Override boolean rename(Path src, Path dst); @SuppressWarnings("deprecation") @Override void rename(Path src, Path dst, final Options.Rename... options); @Override boolean delete(Path f, final boolean recursive); @Override ContentSummary getContentSummary(Path f); void setQuota(Path src, final long namespaceQuota,
      final long diskspaceQuota); @Override FileStatus[] listStatus(Path p); boolean mkdir(Path f, FsPermission permission); @Override boolean mkdirs(Path f, FsPermission permission); @Override void close(); @Override String toString(); @InterfaceAudience.Private @VisibleForTesting DFSClient getClient(); @Override FsStatus getStatus(Path p); @Deprecated DiskStatus getDiskStatus(); @Deprecated long getRawCapacity(); @Deprecated long getRawUsed(); long getMissingBlocksCount(); long getUnderReplicatedBlocksCount(); long getCorruptBlocksCount(); @Override RemoteIterator<Path> listCorruptFileBlocks(Path path); DatanodeInfo[] getDataNodeStats(); DatanodeInfo[] getDataNodeStats(final DatanodeReportType type
      ); boolean setSafeMode(HdfsConstants.SafeModeAction action); boolean setSafeMode(HdfsConstants.SafeModeAction action,
      boolean isChecked); void saveNamespace(); long rollEdits(); boolean restoreFailedStorage(String arg); void refreshNodes(); void finalizeUpgrade(); RollingUpgradeInfo rollingUpgrade(RollingUpgradeAction action); void metaSave(String pathname); @Override FsServerDefaults getServerDefaults(); @Override FileStatus getFileStatus(Path f); @SuppressWarnings("deprecation") @Override void createSymlink(final Path target, final Path link,
      final boolean createParent); @Override boolean supportsSymlinks(); @Override FileStatus getFileLinkStatus(final Path f); @Override Path getLinkTarget(final Path f); @Override FileChecksum getFileChecksum(Path f); @Override FileChecksum getFileChecksum(Path f, final long length); @Override void setPermission(Path p, final FsPermission permission
      ); @Override void setOwner(Path p, final String username, final String groupname
      ); @Override void setTimes(Path p, final long mtime, final long atime
      ); @Override Token<DelegationTokenIdentifier> getDelegationToken(String renewer); void setBalancerBandwidth(long bandwidth); @Override String getCanonicalServiceName(); boolean isInSafeMode(); void allowSnapshot(final Path path); void disallowSnapshot(final Path path); @Override Path createSnapshot(final Path path, final String snapshotName); @Override void renameSnapshot(final Path path, final String snapshotOldName,
      final String snapshotNewName); SnapshottableDirectoryStatus[] getSnapshottableDirListing(); @Override void deleteSnapshot(final Path snapshotDir, final String snapshotName); SnapshotDiffReport getSnapshotDiffReport(final Path snapshotDir,
      final String fromSnapshot, final String toSnapshot); boolean isFileClosed(final Path src); long addCacheDirective(CacheDirectiveInfo info); long addCacheDirective(
      CacheDirectiveInfo info, EnumSet<CacheFlag> flags); void modifyCacheDirective(CacheDirectiveInfo info); void modifyCacheDirective(
      CacheDirectiveInfo info, EnumSet<CacheFlag> flags); void removeCacheDirective(long id); RemoteIterator<CacheDirectiveEntry> listCacheDirectives(
      CacheDirectiveInfo filter); void addCachePool(CachePoolInfo info); void modifyCachePool(CachePoolInfo info); void removeCachePool(String poolName); RemoteIterator<CachePoolEntry> listCachePools(); @Override void modifyAclEntries(Path path, final List<AclEntry> aclSpec); @Override void removeAclEntries(Path path, final List<AclEntry> aclSpec); @Override void removeDefaultAcl(Path path); @Override void removeAcl(Path path); @Override void setAcl(Path path, final List<AclEntry> aclSpec); @Override AclStatus getAclStatus(Path path); void createEncryptionZone(Path path, String keyName); EncryptionZone getEZForPath(Path path); RemoteIterator<EncryptionZone> listEncryptionZones(); @Override void setXAttr(Path path, final String name, final byte[] value, 
      final EnumSet<XAttrSetFlag> flag); @Override byte[] getXAttr(Path path, final String name); @Override Map<String, byte[]> getXAttrs(Path path); @Override Map<String, byte[]> getXAttrs(Path path, final List<String> names); @Override List<String> listXAttrs(Path path); @Override void removeXAttr(Path path, final String name); @Override void access(Path path, final FsAction mode); @Override Token<?>[] addDelegationTokens(
      final String renewer, Credentials credentials); DFSInotifyEventInputStream getInotifyEventStream(); DFSInotifyEventInputStream getInotifyEventStream(long lastReadTxid);  }