Merge pull request #14 from wucheng/master

1 add file lock cleaner interface 2 add get home direcotry interface

jar/jar.md5

@@ -1 +1 @@
-MD5 (chdfs_hadoop_plugin_network-2.1.jar) = c60e3cdf3b5f3a0b5094ea6cc4392b0f
+MD5 (chdfs_hadoop_plugin_network-2.3.jar) = b165f29ad9266228972603d036595a86

pom.xml

@@ -5,7 +5,7 @@
 
  <groupId>com.qcloud</groupId>
  <artifactId>chdfs_hadoop_plugin_network</artifactId>
- <version>2.1</version>
+ <version>2.3</version>
  <packaging>jar</packaging>
 
  <name>chdfs_hadoop_plugin_network</name>

src/main/java/com/qcloud/chdfs/fs/AlreadyLoadedFileSystemInfo.java

@@ -6,9 +6,9 @@ public class AlreadyLoadedFileSystemInfo {
  String versionId;
  String jarPath;
  String jarMd5;
- FileSystem actualFileSystem;
+ FileSystemWithLockCleaner actualFileSystem;
 
- public AlreadyLoadedFileSystemInfo(String versionId, String jarPath, String jarMd5, FileSystem actualFileSystem) {
+ public AlreadyLoadedFileSystemInfo(String versionId, String jarPath, String jarMd5, FileSystemWithLockCleaner actualFileSystem) {
  this.versionId = versionId;
  this.jarPath = jarPath;
  this.jarMd5 = jarMd5;

src/main/java/com/qcloud/chdfs/fs/CHDFSHadoopFileSystemAdapter.java

@@ -20,7 +20,7 @@
 import java.util.Map;
 import java.util.regex.Pattern;
 
-public class CHDFSHadoopFileSystemAdapter extends FileSystem {
+public class CHDFSHadoopFileSystemAdapter extends FileSystemWithLockCleaner {
  private static final Logger log = LoggerFactory.getLogger(CHDFSHadoopFileSystemAdapter.class);
 
  static final String SCHEME = "ofs";
@@ -34,7 +34,7 @@ public class CHDFSHadoopFileSystemAdapter extends FileSystem {
  private static final int DEFAULT_CHDFS_META_SERVER_PORT = 443;
 
  private CHDFSHadoopFileSystemJarLoader jarLoader = new CHDFSHadoopFileSystemJarLoader();
- private FileSystem actualImplFS = null;
+ private FileSystemWithLockCleaner actualImplFS = null;
  private URI uri = null;
  private Path workingDir = null;
  private long initStartMs;
@@ -334,6 +334,14 @@ public Path getWorkingDirectory() {
  return this.workingDir;
  }
 
+ @Override
+ public Path getHomeDirectory() {
+ if (this.actualImplFS == null) {
+ return super.getHomeDirectory();
+ }
+ return this.actualImplFS.getHomeDirectory();
+ }
+
  @java.lang.Override
  public boolean mkdirs(Path f, FsPermission permission) throws IOException {
  if (this.actualImplFS == null) {
@@ -726,6 +734,21 @@ public ContentSummary getContentSummary(Path f) throws IOException {
  }
  }
 
+ @Override
+ public void releaseFileLock(Path p) throws IOException {
+ if (this.actualImplFS == null) {
+ throw new IOException("please init the fileSystem first!");
+ }
+ try {
+ this.actualImplFS.releaseFileLock(p);
+ } catch (IOException ioe) {
+ throw ioe;
+ } catch (Exception e) {
+ log.error("releaseFileLock failed! a unexpected exception occur!", e);
+ throw new IOException("releaseFileLock failed! a unexpected exception occur! " + e.getMessage());
+ }
+ }
+
  @Override
  public void close() throws IOException {
  if (this.actualImplFS == null) {

src/main/java/com/qcloud/chdfs/fs/CHDFSHadoopFileSystemJarLoader.java

@@ -22,7 +22,7 @@ class CHDFSHadoopFileSystemJarLoader {
  private String jarPath;
  private String jarMd5;
 
- private FileSystem actualFileSystem;
+ private FileSystemWithLockCleaner actualFileSystem;
  private static AlreadyLoadedFileSystemInfo alreadyLoadedFileSystemInfo;
 
  private String chdfsDataTransferEndpointSuffix;
@@ -70,8 +70,8 @@ private boolean queryJarPluginInfo(String mountPointAddr, long appid, int jarPlu
  try {
  HttpURLConnection conn = (HttpURLConnection) queryJarUrl.openConnection();
  conn.setRequestProperty("Connection", "Keep-Alive");
+ conn.setReadTimeout(120000);
  conn.connect();
-
  BufferedInputStream bis = new BufferedInputStream(conn.getInputStream());
  ByteArrayOutputStream bos = new ByteArrayOutputStream();
  byte[] buf = new byte[4096];
@@ -148,15 +148,16 @@ synchronized boolean init(String mountPointAddr, long appid, int jarPluginServer
  return true;
  }
 
- FileSystem getActualFileSystem() {
+ FileSystemWithLockCleaner getActualFileSystem() {
  return actualFileSystem;
  }
 
  public void setChdfsDataTransferEndpointSuffix(String chdfsDataTransferEndpointSuffix) {
  this.chdfsDataTransferEndpointSuffix = chdfsDataTransferEndpointSuffix;
  }
 
- private static synchronized FileSystem getAlreadyLoadedClassInfo(ClassLoader currentClassLoader, String jarPath, String versionId, String jarMd5, String tmpDirPath) {
+ private static synchronized FileSystemWithLockCleaner getAlreadyLoadedClassInfo(ClassLoader currentClassLoader, String jarPath,
+ String versionId, String jarMd5, String tmpDirPath) {
  if (alreadyLoadedFileSystemInfo != null
  && alreadyLoadedFileSystemInfo.jarPath.equals(jarPath)
  && alreadyLoadedFileSystemInfo.versionId.equals(versionId)
@@ -182,7 +183,7 @@ private static synchronized FileSystem getAlreadyLoadedClassInfo(ClassLoader cur
  "chdfs.%s.com.qcloud.chdfs.fs.CHDFSHadoopFileSystem", versionId);
  try {
  Class chdfsFSClass = chdfsJarClassLoader.loadClass(className);
- FileSystem actualFileSystem = (FileSystem) chdfsFSClass.newInstance();
+ FileSystemWithLockCleaner actualFileSystem = (FileSystemWithLockCleaner) chdfsFSClass.newInstance();
  alreadyLoadedFileSystemInfo = new AlreadyLoadedFileSystemInfo(versionId, jarPath, jarMd5, actualFileSystem);
  return actualFileSystem;
  } catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {
@@ -325,5 +326,4 @@ private static File downloadJarPath(String jarPath, String versionId, String jar
  }
  }
  }
-
 }

src/main/java/com/qcloud/chdfs/fs/FileLockCleaner.java

@@ -0,0 +1,9 @@
+package com.qcloud.chdfs.fs;
+
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+
+public interface FileLockCleaner {
+ void releaseFileLock(Path p) throws IOException;
+}

src/main/java/com/qcloud/chdfs/fs/FileSystemWithLockCleaner.java

@@ -0,0 +1,6 @@
+package com.qcloud.chdfs.fs;
+
+import org.apache.hadoop.fs.FileSystem;
+
+public abstract class FileSystemWithLockCleaner extends FileSystem implements FileLockCleaner {
+}