Repository: accumulo Updated Branches: refs/heads/master 82e9f0c56 -> 47cf834cb
ACCUMULO-4013 Fix build problems in master branch Ensure RFile resources are closed properly, without warnings, for ACCUMULO-3913 using AutoCloseable, avoiding reassigning unclosed references, and using a utility method to create an RFile.Reader which is returned. Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/47cf834c Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/47cf834c Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/47cf834c Branch: refs/heads/master Commit: 47cf834cbaceb6975a0b7e2abbce87124bb91c92 Parents: 82e9f0c Author: Christopher Tubbs <[email protected]> Authored: Wed Sep 30 11:19:24 2015 -0400 Committer: Christopher Tubbs <[email protected]> Committed: Wed Sep 30 11:19:24 2015 -0400 ---------------------------------------------------------------------- .../accumulo/core/file/FileSKVIterator.java | 1 + .../accumulo/core/file/FileSKVWriter.java | 3 +- .../accumulo/core/file/rfile/PrintInfo.java | 4 +- .../core/file/rfile/RFileOperations.java | 17 ++++--- .../accumulo/core/file/rfile/SplitLarge.java | 47 ++++++++++---------- 5 files changed, 37 insertions(+), 35 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/accumulo/blob/47cf834c/core/src/main/java/org/apache/accumulo/core/file/FileSKVIterator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/accumulo/core/file/FileSKVIterator.java b/core/src/main/java/org/apache/accumulo/core/file/FileSKVIterator.java index 3713453..364a44d 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/FileSKVIterator.java +++ b/core/src/main/java/org/apache/accumulo/core/file/FileSKVIterator.java @@ -34,5 +34,6 @@ public interface FileSKVIterator extends InterruptibleIterator, AutoCloseable { void closeDeepCopies() throws IOException; + @Override void close() throws IOException; } http://git-wip-us.apache.org/repos/asf/accumulo/blob/47cf834c/core/src/main/java/org/apache/accumulo/core/file/FileSKVWriter.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/accumulo/core/file/FileSKVWriter.java b/core/src/main/java/org/apache/accumulo/core/file/FileSKVWriter.java index 98a366f..eefdc6d 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/FileSKVWriter.java +++ b/core/src/main/java/org/apache/accumulo/core/file/FileSKVWriter.java @@ -24,7 +24,7 @@ import org.apache.accumulo.core.data.ByteSequence; import org.apache.accumulo.core.data.Key; import org.apache.accumulo.core.data.Value; -public interface FileSKVWriter { +public interface FileSKVWriter extends AutoCloseable { boolean supportsLocalityGroups(); void startNewLocalityGroup(String name, Set<ByteSequence> columnFamilies) throws IOException; @@ -35,6 +35,7 @@ public interface FileSKVWriter { DataOutputStream createMetaStore(String name) throws IOException; + @Override void close() throws IOException; long getLength() throws IOException; http://git-wip-us.apache.org/repos/asf/accumulo/blob/47cf834c/core/src/main/java/org/apache/accumulo/core/file/rfile/PrintInfo.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/accumulo/core/file/rfile/PrintInfo.java b/core/src/main/java/org/apache/accumulo/core/file/rfile/PrintInfo.java index 4631a4d..cfe571c 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/rfile/PrintInfo.java +++ b/core/src/main/java/org/apache/accumulo/core/file/rfile/PrintInfo.java @@ -122,7 +122,7 @@ public class PrintInfo implements KeywordExecutable { if (opts.histogram || opts.dump || opts.vis || opts.hash) { localityGroupCF = iter.getLocalityGroupCF(); - FileSKVIterator dataIter = iter; + FileSKVIterator dataIter; if (opts.useSample) { dataIter = iter.getSample(); @@ -130,6 +130,8 @@ public class PrintInfo implements KeywordExecutable { System.out.println("ERROR : This rfile has no sample data"); return; } + } else { + dataIter = iter; } for (Entry<String,ArrayList<ByteSequence>> cf : localityGroupCF.entrySet()) { http://git-wip-us.apache.org/repos/asf/accumulo/blob/47cf834c/core/src/main/java/org/apache/accumulo/core/file/rfile/RFileOperations.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/accumulo/core/file/rfile/RFileOperations.java b/core/src/main/java/org/apache/accumulo/core/file/rfile/RFileOperations.java index 17e8e96..a41785a 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/rfile/RFileOperations.java +++ b/core/src/main/java/org/apache/accumulo/core/file/rfile/RFileOperations.java @@ -51,21 +51,20 @@ public class RFileOperations extends FileOperations { @Override public FileSKVIterator openIndex(String file, FileSystem fs, Configuration conf, AccumuloConfiguration acuconf) throws IOException { - return openIndex(file, fs, conf, acuconf, null, null); } - @Override - public FileSKVIterator openIndex(String file, FileSystem fs, Configuration conf, AccumuloConfiguration acuconf, BlockCache dataCache, BlockCache indexCache) - throws IOException { + private static RFile.Reader getReader(String file, FileSystem fs, Configuration conf, AccumuloConfiguration acuconf, BlockCache dataCache, + BlockCache indexCache) throws IOException { Path path = new Path(file); - // long len = fs.getFileStatus(path).getLen(); - // FSDataInputStream in = fs.open(path); - // Reader reader = new RFile.Reader(in, len , conf); CachableBlockFile.Reader _cbr = new CachableBlockFile.Reader(fs, path, conf, dataCache, indexCache, acuconf); - final Reader reader = new RFile.Reader(_cbr); + return new RFile.Reader(_cbr); + } - return reader.getIndex(); + @Override + public FileSKVIterator openIndex(String file, FileSystem fs, Configuration conf, AccumuloConfiguration acuconf, BlockCache dataCache, BlockCache indexCache) + throws IOException { + return getReader(file, fs, conf, acuconf, indexCache, indexCache).getIndex(); } @Override http://git-wip-us.apache.org/repos/asf/accumulo/blob/47cf834c/core/src/main/java/org/apache/accumulo/core/file/rfile/SplitLarge.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/accumulo/core/file/rfile/SplitLarge.java b/core/src/main/java/org/apache/accumulo/core/file/rfile/SplitLarge.java index 92a9f72..4e5b232 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/rfile/SplitLarge.java +++ b/core/src/main/java/org/apache/accumulo/core/file/rfile/SplitLarge.java @@ -60,35 +60,34 @@ public class SplitLarge { AccumuloConfiguration aconf = DefaultConfiguration.getDefaultConfiguration(); Path path = new Path(file); CachableBlockFile.Reader rdr = new CachableBlockFile.Reader(fs, path, conf, null, null, aconf); - Reader iter = new RFile.Reader(rdr); + try (Reader iter = new RFile.Reader(rdr)) { - if (!file.endsWith(".rf")) { - throw new IllegalArgumentException("File must end with .rf"); - } - String smallName = file.substring(0, file.length() - 3) + "_small.rf"; - String largeName = file.substring(0, file.length() - 3) + "_large.rf"; + if (!file.endsWith(".rf")) { + throw new IllegalArgumentException("File must end with .rf"); + } + String smallName = file.substring(0, file.length() - 3) + "_small.rf"; + String largeName = file.substring(0, file.length() - 3) + "_large.rf"; - int blockSize = (int) aconf.getMemoryInBytes(Property.TABLE_FILE_BLOCK_SIZE); - Writer small = new RFile.Writer(new CachableBlockFile.Writer(fs, new Path(smallName), "gz", conf, aconf), blockSize); - small.startDefaultLocalityGroup(); - Writer large = new RFile.Writer(new CachableBlockFile.Writer(fs, new Path(largeName), "gz", conf, aconf), blockSize); - large.startDefaultLocalityGroup(); + int blockSize = (int) aconf.getMemoryInBytes(Property.TABLE_FILE_BLOCK_SIZE); + try (Writer small = new RFile.Writer(new CachableBlockFile.Writer(fs, new Path(smallName), "gz", conf, aconf), blockSize); + Writer large = new RFile.Writer(new CachableBlockFile.Writer(fs, new Path(largeName), "gz", conf, aconf), blockSize)) { + small.startDefaultLocalityGroup(); + large.startDefaultLocalityGroup(); + + iter.seek(new Range(), new ArrayList<ByteSequence>(), false); + while (iter.hasTop()) { + Key key = iter.getTopKey(); + Value value = iter.getTopValue(); + if (key.getSize() + value.getSize() < opts.maxSize) { + small.append(key, value); + } else { + large.append(key, value); + } + iter.next(); + } - iter.seek(new Range(), new ArrayList<ByteSequence>(), false); - while (iter.hasTop()) { - Key key = iter.getTopKey(); - Value value = iter.getTopValue(); - if (key.getSize() + value.getSize() < opts.maxSize) { - small.append(key, value); - } else { - large.append(key, value); } - iter.next(); } - - iter.close(); - large.close(); - small.close(); } }
