Further improved the HdfsResource by managing filesystem.close() better Project: http://git-wip-us.apache.org/repos/asf/metamodel/repo Commit: http://git-wip-us.apache.org/repos/asf/metamodel/commit/08e3a8f9 Tree: http://git-wip-us.apache.org/repos/asf/metamodel/tree/08e3a8f9 Diff: http://git-wip-us.apache.org/repos/asf/metamodel/diff/08e3a8f9
Branch: refs/heads/master Commit: 08e3a8f9cda7b689f4ed0bafb9ad113276ca41d0 Parents: 4621a28 Author: Kasper Sørensen <[email protected]> Authored: Sun Jun 14 20:58:08 2015 +0200 Committer: Kasper Sørensen <[email protected]> Committed: Sun Jun 14 20:58:08 2015 +0200 ---------------------------------------------------------------------- .../org/apache/metamodel/util/HdfsResource.java | 167 +++++++++++++------ .../util/HdfsResourceIntegrationTest.java | 3 - hadoop/src/test/resources/log4j.xml | 21 +++ 3 files changed, 135 insertions(+), 56 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/metamodel/blob/08e3a8f9/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java ---------------------------------------------------------------------- diff --git a/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java b/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java index 4590bde..c2f38ce 100644 --- a/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java +++ b/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java @@ -26,6 +26,7 @@ import java.util.regex.Matcher; import java.util.regex.Pattern; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.metamodel.MetaModelException; @@ -91,108 +92,168 @@ public class HdfsResource implements Resource { @Override public boolean isExists() { - return doWithFileSystem(new UncheckedFunc<FileSystem, Boolean>() { - @Override - protected Boolean evalUnchecked(FileSystem fs) throws Exception { - return fs.exists(getHadoopPath()); - } - }); + final FileSystem fs = getHadoopFileSystem(); + try { + return fs.exists(getHadoopPath()); + } catch (Exception e) { + throw wrapException(e); + } finally { + FileHelper.safeClose(fs); + } } @Override public long getSize() { - return doWithFileSystem(new UncheckedFunc<FileSystem, Long>() { - @Override - protected Long evalUnchecked(FileSystem fs) throws Exception { - return fs.getFileStatus(getHadoopPath()).getLen(); - } - }); + final FileSystem fs = getHadoopFileSystem(); + try { + return fs.getFileStatus(getHadoopPath()).getLen(); + } catch (Exception e) { + throw wrapException(e); + } finally { + FileHelper.safeClose(fs); + } } @Override public long getLastModified() { - return doWithFileSystem(new UncheckedFunc<FileSystem, Long>() { - @Override - protected Long evalUnchecked(FileSystem fs) throws Exception { - return fs.getFileStatus(getHadoopPath()).getModificationTime(); - } - }); + final FileSystem fs = getHadoopFileSystem(); + try { + return fs.getFileStatus(getHadoopPath()).getModificationTime(); + } catch (Exception e) { + throw wrapException(e); + } finally { + FileHelper.safeClose(fs); + } } @Override public void write(final Action<OutputStream> writeCallback) throws ResourceException { - final OutputStream out = doWithFileSystem(new UncheckedFunc<FileSystem, OutputStream>() { - @Override - protected OutputStream evalUnchecked(FileSystem fs) throws Exception { - return fs.create(getHadoopPath(), true); - } - }); + final FileSystem fs = getHadoopFileSystem(); try { - writeCallback.run(out); + final FSDataOutputStream out = fs.create(getHadoopPath(), true); + try { + writeCallback.run(out); + } finally { + FileHelper.safeClose(out); + } } catch (Exception e) { throw wrapException(e); } finally { - FileHelper.safeClose(out); + FileHelper.safeClose(fs); } } @Override public void append(Action<OutputStream> appendCallback) throws ResourceException { - final OutputStream out = doWithFileSystem(new UncheckedFunc<FileSystem, OutputStream>() { - @Override - protected OutputStream evalUnchecked(FileSystem fs) throws Exception { - return fs.append(getHadoopPath()); - } - }); + final FileSystem fs = getHadoopFileSystem(); try { - appendCallback.run(out); + final FSDataOutputStream out = fs.append(getHadoopPath()); + try { + appendCallback.run(out); + } finally { + FileHelper.safeClose(out); + } } catch (Exception e) { throw wrapException(e); } finally { - FileHelper.safeClose(out); + FileHelper.safeClose(fs); } } @Override public InputStream read() throws ResourceException { - return doWithFileSystem(new UncheckedFunc<FileSystem, InputStream>() { + final FileSystem fs = getHadoopFileSystem(); + final InputStream in; + try { + in = fs.open(getHadoopPath()); + } catch (Exception e) { + // we can close 'fs' in case of an exception + FileHelper.safeClose(fs); + throw wrapException(e); + } + + // return a wrappper InputStream which manages the 'fs' closeable + return new InputStream() { @Override - protected InputStream evalUnchecked(FileSystem fs) throws Exception { - return fs.open(getHadoopPath()); + public int read() throws IOException { + return in.read(); } - }); + + @Override + public int read(byte[] b, int off, int len) throws IOException { + return in.read(b, off, len); + } + + @Override + public int read(byte[] b) throws IOException { + return in.read(b); + } + + @Override + public boolean markSupported() { + return in.markSupported(); + } + + @Override + public synchronized void mark(int readlimit) { + in.mark(readlimit); + } + + @Override + public int available() throws IOException { + return in.available(); + } + + @Override + public synchronized void reset() throws IOException { + in.reset(); + } + + @Override + public long skip(long n) throws IOException { + return in.skip(n); + } + + @Override + public void close() throws IOException { + super.close(); + // need to close 'fs' when input stream is closed + FileHelper.safeClose(fs); + } + }; } @Override public void read(Action<InputStream> readCallback) throws ResourceException { - final InputStream in = read(); + final FileSystem fs = getHadoopFileSystem(); try { - readCallback.run(in); + final InputStream in = fs.open(getHadoopPath()); + try { + readCallback.run(in); + } finally { + FileHelper.safeClose(in); + } } catch (Exception e) { throw wrapException(e); } finally { - FileHelper.safeClose(in); + FileHelper.safeClose(fs); } } @Override public <E> E read(Func<InputStream, E> readCallback) throws ResourceException { - final InputStream in = read(); + final FileSystem fs = getHadoopFileSystem(); try { - return readCallback.eval(in); + final InputStream in = fs.open(getHadoopPath()); + try { + return readCallback.eval(in); + } finally { + FileHelper.safeClose(in); + } } catch (Exception e) { throw wrapException(e); } finally { - FileHelper.safeClose(in); - } - } - - private <E> E doWithFileSystem(Func<FileSystem, E> action) { - final FileSystem hadoopFileSystem = getHadoopFileSystem(); - try { - return action.eval(hadoopFileSystem); - } catch (Exception e) { - throw wrapException(e); + FileHelper.safeClose(fs); } } http://git-wip-us.apache.org/repos/asf/metamodel/blob/08e3a8f9/hadoop/src/test/java/org/apache/metamodel/util/HdfsResourceIntegrationTest.java ---------------------------------------------------------------------- diff --git a/hadoop/src/test/java/org/apache/metamodel/util/HdfsResourceIntegrationTest.java b/hadoop/src/test/java/org/apache/metamodel/util/HdfsResourceIntegrationTest.java index 5ef5e53..3f37379 100644 --- a/hadoop/src/test/java/org/apache/metamodel/util/HdfsResourceIntegrationTest.java +++ b/hadoop/src/test/java/org/apache/metamodel/util/HdfsResourceIntegrationTest.java @@ -80,9 +80,6 @@ public class HdfsResourceIntegrationTest extends TestCase { final HdfsResource res1 = new HdfsResource(_hostname, _port, _filePath); logger.info(stopwatch.elapsed(TimeUnit.MILLISECONDS) + " - start"); - assertFalse(res1.isExists()); - logger.info(stopwatch.elapsed(TimeUnit.MILLISECONDS) + " - exists"); - res1.write(new Action<OutputStream>() { @Override public void run(OutputStream out) throws Exception { http://git-wip-us.apache.org/repos/asf/metamodel/blob/08e3a8f9/hadoop/src/test/resources/log4j.xml ---------------------------------------------------------------------- diff --git a/hadoop/src/test/resources/log4j.xml b/hadoop/src/test/resources/log4j.xml new file mode 100644 index 0000000..29f497f --- /dev/null +++ b/hadoop/src/test/resources/log4j.xml @@ -0,0 +1,21 @@ +<?xml version="1.0" encoding="UTF-8" ?> +<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd"> +<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/";> + + <appender name="consoleAppender" class="org.apache.log4j.ConsoleAppender"> + <param name="Target" value="System.out" /> + <layout class="org.apache.log4j.PatternLayout"> + <param name="ConversionPattern" value="%-5p %d{HH:mm:ss} %c{1} - %m%n" /> + </layout> + </appender> + + <logger name="org.apache.metamodel"> + <level value="info" /> + </logger> + + <root> + <priority value="fatal" /> + <appender-ref ref="consoleAppender" /> + </root> + +</log4j:configuration> \ No newline at end of file
