Adding logic for getting import status from table stats.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/99aa2408 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/99aa2408 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/99aa2408 Branch: refs/heads/master Commit: 99aa2408599aba47eee9d5049706b50288fdc889 Parents: 55c3abb Author: Aaron McCurry <[email protected]> Authored: Sat Mar 28 11:10:31 2015 -0400 Committer: Aaron McCurry <[email protected]> Committed: Sat Mar 28 11:10:31 2015 -0400 ---------------------------------------------------------------------- .../org/apache/blur/manager/IndexServer.java | 5 +- .../indexserver/AbstractIndexServer.java | 20 +++++ .../indexserver/DistributedIndexServer.java | 2 + .../manager/indexserver/LocalIndexServer.java | 10 +++ .../blur/manager/stats/MergerTableStats.java | 2 + .../apache/blur/manager/writer/BlurIndex.java | 4 + .../blur/manager/writer/BlurIndexReadOnly.java | 10 +++ .../manager/writer/BlurIndexSimpleWriter.java | 16 ++++ .../blur/manager/writer/IndexImporter.java | 63 ++++++++++++++ .../blur/manager/writer/TestingSeqSorting.java | 92 ++++++++++++++++++++ .../org/apache/blur/thrift/BlurShardServer.java | 2 + .../blur/command/ShardCommandManagerTest.java | 20 +++++ .../blur/manager/writer/IndexImporterTest.java | 10 +++ .../server/cache/ThriftCacheServerTest.java | 10 +++ .../lib/BlurOutputFormatMiniClusterTest.java | 2 +- 15 files changed, 266 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/main/java/org/apache/blur/manager/IndexServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/IndexServer.java b/blur-core/src/main/java/org/apache/blur/manager/IndexServer.java index 9275df4..da28fcb 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/IndexServer.java +++ b/blur-core/src/main/java/org/apache/blur/manager/IndexServer.java @@ -18,7 +18,6 @@ package org.apache.blur.manager; */ import java.io.Closeable; import java.io.IOException; -import java.util.List; import java.util.Map; import java.util.SortedSet; @@ -110,4 +109,8 @@ public interface IndexServer extends Closeable { */ Map<String, ShardState> getShardState(String table); + long getSegmentImportInProgressCount(String table) throws IOException; + + long getSegmentImportPendingCount(String table) throws IOException; + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/main/java/org/apache/blur/manager/indexserver/AbstractIndexServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/AbstractIndexServer.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/AbstractIndexServer.java index 50f7df8..cc0fa9d 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/AbstractIndexServer.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/AbstractIndexServer.java @@ -44,4 +44,24 @@ public abstract class AbstractIndexServer implements IndexServer { return rowCount; } + @Override + public long getSegmentImportInProgressCount(String table) throws IOException { + long segmentImportInProgressCount = 0; + Map<String, BlurIndex> indexes = getIndexes(table); + for (Map.Entry<String, BlurIndex> index : indexes.entrySet()) { + segmentImportInProgressCount += index.getValue().getSegmentImportInProgressCount(); + } + return segmentImportInProgressCount; + } + + @Override + public long getSegmentImportPendingCount(String table) throws IOException { + long segmentImportPendingCount = 0; + Map<String, BlurIndex> indexes = getIndexes(table); + for (Map.Entry<String, BlurIndex> index : indexes.entrySet()) { + segmentImportPendingCount += index.getValue().getSegmentImportPendingCount(); + } + return segmentImportPendingCount; + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java index 2f9ccba..addc1b0 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java @@ -660,4 +660,6 @@ public class DistributedIndexServer extends AbstractDistributedIndexServer { throw new RuntimeException(e); } } + + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java b/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java index ba7987b..fde7b35 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java +++ b/blur-core/src/main/java/org/apache/blur/manager/indexserver/LocalIndexServer.java @@ -203,4 +203,14 @@ public class LocalIndexServer extends AbstractIndexServer { public Map<String, ShardState> getShardState(String table) { throw new RuntimeException("Not supported yet."); } + + @Override + public long getSegmentImportInProgressCount(String table) { + return 0l; + } + + @Override + public long getSegmentImportPendingCount(String table) { + return 0l; + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/main/java/org/apache/blur/manager/stats/MergerTableStats.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/stats/MergerTableStats.java b/blur-core/src/main/java/org/apache/blur/manager/stats/MergerTableStats.java index 9d5eae8..2f89c9e 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/stats/MergerTableStats.java +++ b/blur-core/src/main/java/org/apache/blur/manager/stats/MergerTableStats.java @@ -49,6 +49,8 @@ public class MergerTableStats implements Merger<TableStats> { s1.bytes = Math.max(s1.bytes, s2.bytes); s1.recordCount = s1.recordCount + s2.recordCount; s1.rowCount = s1.rowCount + s2.rowCount; + s1.segmentImportInProgressCount = s1.segmentImportInProgressCount + s2.segmentImportInProgressCount; + s1.segmentImportPendingCount = s1.segmentImportPendingCount + s2.segmentImportPendingCount; return s1; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java index cf1865f..684f09b 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndex.java @@ -141,4 +141,8 @@ public abstract class BlurIndex { public abstract void addBulkMutate(String bulkId, RowMutation mutation) throws IOException; + public abstract long getSegmentImportPendingCount() throws IOException; + + public abstract long getSegmentImportInProgressCount() throws IOException; + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReadOnly.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReadOnly.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReadOnly.java index aca14c9..1c66e23 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReadOnly.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexReadOnly.java @@ -84,4 +84,14 @@ public class BlurIndexReadOnly extends BlurIndex { throw new RuntimeException("Read-only shard"); } + @Override + public long getSegmentImportPendingCount() throws IOException { + return _blurIndex.getSegmentImportPendingCount(); + } + + @Override + public long getSegmentImportInProgressCount() throws IOException { + return _blurIndex.getSegmentImportInProgressCount(); + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java index 7ea3634..65b0beb 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/BlurIndexSimpleWriter.java @@ -978,4 +978,20 @@ public class BlurIndexSimpleWriter extends BlurIndex { return false; } } + + @Override + public long getSegmentImportPendingCount() throws IOException { + if (_indexImporter != null) { + return _indexImporter.getSegmentImportPendingCount(); + } + return 0l; + } + + @Override + public long getSegmentImportInProgressCount() throws IOException { + if (_indexImporter != null) { + return _indexImporter.getSegmentImportInProgressCount(); + } + return 0l; + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java b/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java index 093c583..99c4f71 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java @@ -99,6 +99,69 @@ public class IndexImporter extends TimerTask implements Closeable { _inindexImporterTimer.purge(); } + public long getSegmentImportPendingCount() throws IOException { + Path path = _shardContext.getHdfsDirPath(); + Configuration configuration = _shardContext.getTableContext().getConfiguration(); + FileSystem fileSystem = path.getFileSystem(configuration); + for (int i = 0; i < 10; i++) { + try { + FileStatus[] listStatus = fileSystem.listStatus(path, new PathFilter() { + @Override + public boolean accept(Path path) { + if (path != null && path.getName().endsWith(COMMIT)) { + return true; + } + return false; + } + }); + return listStatus.length; + } catch (FileNotFoundException e) { + LOG.warn("File not found error, retrying."); + } + try { + Thread.sleep(100); + } catch (InterruptedException e) { + return 0L; + } + } + throw new IOException("Received too many errors. Give up."); + } + + public long getSegmentImportInProgressCount() throws IOException { + Path path = _shardContext.getHdfsDirPath(); + Configuration configuration = _shardContext.getTableContext().getConfiguration(); + FileSystem fileSystem = path.getFileSystem(configuration); + for (int i = 0; i < 10; i++) { + try { + FileStatus[] listStatus = fileSystem.listStatus(path, new PathFilter() { + @Override + public boolean accept(Path path) { + if (path != null && path.getName().endsWith(INUSE)) { + return true; + } + return false; + } + }); + long count = 0; + for (FileStatus fileStatus : listStatus) { + Path p = fileStatus.getPath(); + if (fileSystem.exists(new Path(p, INPROGRESS))) { + count++; + } + } + return count; + } catch (FileNotFoundException e) { + LOG.warn("File not found error, retrying."); + } + try { + Thread.sleep(100); + } catch (InterruptedException e) { + return 0L; + } + } + throw new IOException("Received too many errors. Give up."); + } + @Override public void run() { // Only allow one import to occur in the process at a time. http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/main/java/org/apache/blur/manager/writer/TestingSeqSorting.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/TestingSeqSorting.java b/blur-core/src/main/java/org/apache/blur/manager/writer/TestingSeqSorting.java new file mode 100644 index 0000000..5e577ba --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/TestingSeqSorting.java @@ -0,0 +1,92 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.blur.manager.writer; + +import java.io.IOException; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.BytesWritable; +import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.io.SequenceFile; +import org.apache.hadoop.io.SequenceFile.Sorter; + +public class TestingSeqSorting { + + public static void main(String[] args) throws IOException { + Configuration conf = new Configuration(); + Path p = new Path("hdfs://localhost:9000/testsort/data.seq"); + // URI uri = new File(".").getAbsoluteFile().toURI(); + // Path local = new Path(uri); + + // Path p = new Path(new Path(local, "tmp"), "data.seq"); + // { + // Writer writer = SequenceFile.createWriter(conf, + // Writer.valueClass(LongWritable.class), + // Writer.keyClass(BytesWritable.class), Writer.file(p)); + // + // LongWritable value = new LongWritable(); + // BytesWritable key = new BytesWritable(); + // Random random = new Random(1); + // byte[] buf = new byte[50]; + // for (int i = 0; i < 10000000; i++) { + // value.set(random.nextLong()); + // random.nextBytes(buf); + // key.set(buf, 0, buf.length); + // writer.append(key, value); + // } + // writer.close(); + // } + + // { + // // io.seqfile.local.dir + // URI uri = new File(".").getAbsoluteFile().toURI(); + // Path local = new Path(uri); + // + // conf.set("io.seqfile.local.dir", new Path(local, "tmp").toString()); + // Path tempDir = new Path(p.getParent(), "tmp"); + // Sorter sorter = new SequenceFile.Sorter(p.getFileSystem(conf), + // BytesWritable.class, LongWritable.class, conf); + // sorter.setMemory(10 * 1024 * 1024); + // // Path tempDir = new Path(p.getParent(), "tmp"); + // + // RawKeyValueIterator iterate = sorter.sortAndIterate(new Path[] { p }, + // tempDir, false); + // Path sortedPath = new Path(p.getParent(), "sorted.seq"); + // Writer writer = SequenceFile.createWriter(conf, + // Writer.valueClass(LongWritable.class), + // Writer.keyClass(BytesWritable.class), Writer.file(sortedPath)); + // while (iterate.next()) { + // DataOutputBuffer key = iterate.getKey(); + // byte[] keyData = key.getData(); + // int keyOffset = 0; + // int keyLength = key.getLength(); + // ValueBytes val = iterate.getValue(); + // writer.appendRaw(keyData, keyOffset, keyLength, val); + // } + // writer.close(); + // } + { + conf.setInt("io.sort.factor", 1000); + Sorter sorter = new SequenceFile.Sorter(p.getFileSystem(conf), BytesWritable.class, LongWritable.class, conf); + sorter.setMemory(10 * 1024 * 1024); + sorter.sort(new Path[] { p }, new Path(p.getParent(), "sorted.seq"), false); + } + + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/main/java/org/apache/blur/thrift/BlurShardServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/BlurShardServer.java b/blur-core/src/main/java/org/apache/blur/thrift/BlurShardServer.java index 05fdb7d..b993eb6 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/BlurShardServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/BlurShardServer.java @@ -251,6 +251,8 @@ public class BlurShardServer extends TableAdmin implements Iface { tableStats.recordCount = _indexServer.getRecordCount(table); tableStats.rowCount = _indexServer.getRowCount(table); tableStats.bytes = _indexServer.getTableSize(table); + tableStats.segmentImportInProgressCount = _indexServer.getSegmentImportInProgressCount(table); + tableStats.segmentImportPendingCount = _indexServer.getSegmentImportPendingCount(table); return tableStats; } catch (Exception e) { LOG.error("Unknown error while trying to get table stats [table={0}]", e, table); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/test/java/org/apache/blur/command/ShardCommandManagerTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/command/ShardCommandManagerTest.java b/blur-core/src/test/java/org/apache/blur/command/ShardCommandManagerTest.java index ef41c4b..c6d90bd 100644 --- a/blur-core/src/test/java/org/apache/blur/command/ShardCommandManagerTest.java +++ b/blur-core/src/test/java/org/apache/blur/command/ShardCommandManagerTest.java @@ -361,6 +361,16 @@ public class ShardCommandManagerTest { public void close() throws IOException { throw new RuntimeException("Not implemented."); } + + @Override + public long getSegmentImportInProgressCount(String table) throws IOException { + throw new RuntimeException("Not implemented."); + } + + @Override + public long getSegmentImportPendingCount(String table) throws IOException { + throw new RuntimeException("Not implemented."); + } }; } @@ -428,6 +438,16 @@ public class ShardCommandManagerTest { public void addBulkMutate(String bulkId, RowMutation mutation) throws IOException { throw new RuntimeException("Not implemented."); } + + @Override + public long getSegmentImportPendingCount() throws IOException { + throw new RuntimeException("Not implemented."); + } + + @Override + public long getSegmentImportInProgressCount() throws IOException { + throw new RuntimeException("Not implemented."); + } }; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java b/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java index 1cf1e5a..ee48ecd 100644 --- a/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java +++ b/blur-core/src/test/java/org/apache/blur/manager/writer/IndexImporterTest.java @@ -197,6 +197,16 @@ public class IndexImporterTest { public void addBulkMutate(String bulkId, RowMutation mutation) throws IOException { throw new RuntimeException("Not implemented."); } + + @Override + public long getSegmentImportPendingCount() throws IOException { + throw new RuntimeException("Not implemented."); + } + + @Override + public long getSegmentImportInProgressCount() throws IOException { + throw new RuntimeException("Not implemented."); + } }; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-core/src/test/java/org/apache/blur/server/cache/ThriftCacheServerTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/server/cache/ThriftCacheServerTest.java b/blur-core/src/test/java/org/apache/blur/server/cache/ThriftCacheServerTest.java index f01a665..bd43b45 100644 --- a/blur-core/src/test/java/org/apache/blur/server/cache/ThriftCacheServerTest.java +++ b/blur-core/src/test/java/org/apache/blur/server/cache/ThriftCacheServerTest.java @@ -563,6 +563,16 @@ public class ThriftCacheServerTest { public void close() throws IOException { throw new RuntimeException("Not implemented."); } + + @Override + public long getSegmentImportInProgressCount(String table) throws IOException { + throw new RuntimeException("Not implemented."); + } + + @Override + public long getSegmentImportPendingCount(String table) throws IOException { + throw new RuntimeException("Not implemented."); + } }; } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/99aa2408/blur-mapred-hadoop2/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatMiniClusterTest.java ---------------------------------------------------------------------- diff --git a/blur-mapred-hadoop2/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatMiniClusterTest.java b/blur-mapred-hadoop2/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatMiniClusterTest.java index 621a364..5fdc9ab 100644 --- a/blur-mapred-hadoop2/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatMiniClusterTest.java +++ b/blur-mapred-hadoop2/src/test/java/org/apache/blur/mapreduce/lib/BlurOutputFormatMiniClusterTest.java @@ -189,7 +189,7 @@ public class BlurOutputFormatMiniClusterTest { if (tableStats.getRowCount() > 0) { break; } - Thread.sleep(5000); + Thread.sleep(100); } assertTrue(fileSystem.exists(tablePath));
