Adding a new feature to allow for shards to pull directly from a queue like interface.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/052c131e Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/052c131e Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/052c131e Branch: refs/heads/apache-blur-0.2 Commit: 052c131e92e0caefb0c513fe52098ad6c6e04d3a Parents: 31f23a3 Author: Aaron McCurry <[email protected]> Authored: Sat Feb 22 21:47:25 2014 -0500 Committer: Aaron McCurry <[email protected]> Committed: Sat Feb 22 21:47:25 2014 -0500 ---------------------------------------------------------------------- .../org/apache/blur/manager/IndexManager.java | 49 +----- .../manager/writer/BlurIndexSimpleWriter.java | 5 +- .../blur/manager/writer/MutatableAction.java | 53 +++++++ .../apache/blur/manager/writer/QueueReader.java | 125 +++++++++++++++ .../org/apache/blur/server/TableContext.java | 33 ++++ .../writer/BlurIndexSimpleWriterTest.java | 3 - .../writer/QueueReaderBasicInMemory.java | 49 ++++++ .../blur/manager/writer/QueueReaderTest.java | 158 +++++++++++++++++++ .../org/apache/blur/utils/BlurConstants.java | 4 + 9 files changed, 429 insertions(+), 50 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/052c131e/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java b/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java index 093cd81..c8822d0 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java +++ b/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java @@ -460,7 +460,8 @@ public class IndexManager { } String rowId = blurQuery.getRowId(); if (rowId != null) { - // reduce the index selection down to the only one that would contain the row. + // reduce the index selection down to the only one that would contain + // the row. Map<String, BlurIndex> map = new HashMap<String, BlurIndex>(); String shard = MutationHelper.getShardName(table, rowId, getNumberOfShards(table), _blurPartitioner); BlurIndex index = getBlurIndex(table, shard); @@ -1207,25 +1208,7 @@ public class IndexManager { } ShardContext shardContext = blurIndex.getShardContext(); final MutatableAction mutatableAction = new MutatableAction(shardContext); - for (int i = 0; i < mutations.size(); i++) { - RowMutation mutation = mutations.get(i); - RowMutationType type = mutation.rowMutationType; - switch (type) { - case REPLACE_ROW: - Row row = MutationHelper.getRowFromMutations(mutation.rowId, mutation.recordMutations); - mutatableAction.replaceRow(row); - break; - case UPDATE_ROW: - doUpdateRowMutation(mutation, mutatableAction); - break; - case DELETE_ROW: - mutatableAction.deleteRow(mutation.rowId); - break; - default: - throw new RuntimeException("Not supported [" + type + "]"); - } - } - + mutatableAction.mutate(mutations); return _mutateExecutor.submit(new Callable<Void>() { @Override public Void call() throws Exception { @@ -1253,32 +1236,6 @@ public class IndexManager { return map; } - private void doUpdateRowMutation(RowMutation mutation, MutatableAction mutatableAction) throws BlurException, - IOException { - String rowId = mutation.getRowId(); - - for (RecordMutation recordMutation : mutation.getRecordMutations()) { - RecordMutationType type = recordMutation.recordMutationType; - Record record = recordMutation.getRecord(); - switch (type) { - case DELETE_ENTIRE_RECORD: - mutatableAction.deleteRecord(rowId, record.getRecordId()); - break; - case APPEND_COLUMN_VALUES: - mutatableAction.appendColumns(rowId, record); - break; - case REPLACE_ENTIRE_RECORD: - mutatableAction.replaceRecord(rowId, record); - break; - case REPLACE_COLUMNS: - mutatableAction.replaceColumns(rowId, record); - break; - default: - throw new RuntimeException("Unsupported record mutation type [" + type + "]"); - } - } - } - private int getNumberOfShards(String table) { return getTableContext(table).getDescriptor().getShardCount(); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/052c131e/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 f9dc1c8..306479b 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 @@ -67,6 +67,7 @@ public class BlurIndexSimpleWriter extends BlurIndex { private final AtomicReference<BlurIndexWriter> _writer = new AtomicReference<BlurIndexWriter>(); private final boolean _makeReaderExitable = true; private IndexImporter _indexImporter; + private QueueReader _queueReader; private final ReadWriteLock _lock = new ReentrantReadWriteLock(); private final Lock _writeLock = _lock.writeLock(); private final ReadWriteLock _indexRefreshLock = new ReentrantReadWriteLock(); @@ -134,6 +135,7 @@ public class BlurIndexSimpleWriter extends BlurIndex { private Thread getWriterOpener(ShardContext shardContext) { Thread thread = new Thread(new Runnable() { + @Override public void run() { try { @@ -142,6 +144,7 @@ public class BlurIndexSimpleWriter extends BlurIndex { _writer.notify(); } _indexImporter = new IndexImporter(BlurIndexSimpleWriter.this, _shardContext, TimeUnit.SECONDS, 10); + _queueReader = _tableContext.getQueueReader(BlurIndexSimpleWriter.this, _shardContext); } catch (IOException e) { LOG.error("Unknown error on index writer open.", e); } @@ -206,7 +209,7 @@ public class BlurIndexSimpleWriter extends BlurIndex { @Override public void close() throws IOException { _isClosed.set(true); - IOUtils.cleanup(LOG, _indexImporter, _writer.get(), _indexReader.get()); + IOUtils.cleanup(LOG, _indexImporter, _queueReader, _writer.get(), _indexReader.get()); } @Override http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/052c131e/blur-core/src/main/java/org/apache/blur/manager/writer/MutatableAction.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/MutatableAction.java b/blur-core/src/main/java/org/apache/blur/manager/writer/MutatableAction.java index c8f0dfd..2a7159b 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/MutatableAction.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/MutatableAction.java @@ -33,11 +33,16 @@ import org.apache.blur.manager.IndexManager; import org.apache.blur.server.IndexSearcherClosable; import org.apache.blur.server.ShardContext; import org.apache.blur.server.TableContext; +import org.apache.blur.thrift.MutationHelper; import org.apache.blur.thrift.generated.Column; import org.apache.blur.thrift.generated.FetchResult; import org.apache.blur.thrift.generated.FetchRowResult; import org.apache.blur.thrift.generated.Record; +import org.apache.blur.thrift.generated.RecordMutation; +import org.apache.blur.thrift.generated.RecordMutationType; import org.apache.blur.thrift.generated.Row; +import org.apache.blur.thrift.generated.RowMutation; +import org.apache.blur.thrift.generated.RowMutationType; import org.apache.blur.thrift.generated.Selector; import org.apache.blur.utils.BlurConstants; import org.apache.blur.utils.RowDocumentUtil; @@ -376,4 +381,52 @@ public class MutatableAction extends IndexAction { } + public void mutate(RowMutation mutation) { + RowMutationType type = mutation.rowMutationType; + switch (type) { + case REPLACE_ROW: + Row row = MutationHelper.getRowFromMutations(mutation.rowId, mutation.recordMutations); + replaceRow(row); + break; + case UPDATE_ROW: + doUpdateRowMutation(mutation, this); + break; + case DELETE_ROW: + deleteRow(mutation.rowId); + break; + default: + throw new RuntimeException("Not supported [" + type + "]"); + } + } + + private void doUpdateRowMutation(RowMutation mutation, MutatableAction mutatableAction) { + String rowId = mutation.getRowId(); + for (RecordMutation recordMutation : mutation.getRecordMutations()) { + RecordMutationType type = recordMutation.recordMutationType; + Record record = recordMutation.getRecord(); + switch (type) { + case DELETE_ENTIRE_RECORD: + mutatableAction.deleteRecord(rowId, record.getRecordId()); + break; + case APPEND_COLUMN_VALUES: + mutatableAction.appendColumns(rowId, record); + break; + case REPLACE_ENTIRE_RECORD: + mutatableAction.replaceRecord(rowId, record); + break; + case REPLACE_COLUMNS: + mutatableAction.replaceColumns(rowId, record); + break; + default: + throw new RuntimeException("Unsupported record mutation type [" + type + "]"); + } + } + } + + public void mutate(List<RowMutation> mutations) { + for (int i = 0; i < mutations.size(); i++) { + mutate(mutations.get(i)); + } + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/052c131e/blur-core/src/main/java/org/apache/blur/manager/writer/QueueReader.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/QueueReader.java b/blur-core/src/main/java/org/apache/blur/manager/writer/QueueReader.java new file mode 100644 index 0000000..f44dea9 --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/QueueReader.java @@ -0,0 +1,125 @@ +/** + * 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 static org.apache.blur.utils.BlurConstants.BLUR_SHARD_INDEX_QUEUE_READER_BACKOFF; +import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_INDEX_QUEUE_READER_MAX; + +import java.io.Closeable; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.blur.BlurConfiguration; +import org.apache.blur.log.Log; +import org.apache.blur.log.LogFactory; +import org.apache.blur.server.ShardContext; +import org.apache.blur.server.TableContext; +import org.apache.blur.thrift.generated.RowMutation; + +public abstract class QueueReader implements Closeable, Runnable { + + private static final Log LOG = LogFactory.getLog(QueueReader.class); + + protected final ShardContext _shardContext; + protected final BlurIndex _index; + protected final long _backOff; + protected final Thread _daemon; + protected final AtomicBoolean _running = new AtomicBoolean(); + protected final int _max; + protected final TableContext _tableContext; + + public QueueReader(BlurIndex index, ShardContext shardContext) { + _running.set(true); + _index = index; + _shardContext = shardContext; + _tableContext = _shardContext.getTableContext(); + BlurConfiguration configuration = _tableContext.getBlurConfiguration(); + _backOff = configuration.getLong(BLUR_SHARD_INDEX_QUEUE_READER_BACKOFF, 500); + _max = configuration.getInt(BLUR_SHARD_INDEX_QUEUE_READER_MAX, 500); + _daemon = new Thread(this); + _daemon.setName("Queue Loader for [" + _tableContext.getTable() + "/" + shardContext.getShard() + "]"); + _daemon.setDaemon(true); + _daemon.start(); + } + + @Override + public void run() { + List<RowMutation> mutations = new ArrayList<RowMutation>(); + while (_running.get()) { + take(mutations, _max); + if (mutations.isEmpty()) { + try { + Thread.sleep(_backOff); + } catch (InterruptedException e) { + return; + } + } else { + MutatableAction mutatableAction = new MutatableAction(_shardContext); + mutatableAction.mutate(mutations); + try { + _index.process(mutatableAction); + success(); + } catch (IOException e) { + failure(); + LOG.error("Unknown error during loading of rowmutations from queue [{0}] into table [{1}] and shard [{2}].", + this.toString(), _tableContext.getTable(), _shardContext.getShard()); + } finally { + mutations.clear(); + } + } + } + } + + @Override + public void close() throws IOException { + if (_running.get()) { + _running.set(false); + _daemon.interrupt(); + } + } + + /** + * Takes up to the max number of {@link RowMutation}s off the queue and + * returns. The implementation can choose to block until new items are + * available. However if the method returns without adding any items to the + * mutations list, the loading thread will back off a configurable amount of + * time. <br/> + * <br/> + * Configuration setting: "blur.shard.index.queue.reader.backoff" + * + * @param mutations + * @param max + */ + public abstract void take(List<RowMutation> mutations, int max); + + /** + * This method will be called after each successful load of data from the + * queue. This will allow the queue to be notified that the information has + * been successfully loaded. + */ + public abstract void success(); + + /** + * This method will be called after each failed load of data from the queue. + * This will allow the queue to be notified that the information has been WAS + * NOT successfully loaded. + */ + public abstract void failure(); + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/052c131e/blur-core/src/main/java/org/apache/blur/server/TableContext.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/server/TableContext.java b/blur-core/src/main/java/org/apache/blur/server/TableContext.java index 788d34f..8c1f012 100644 --- a/blur-core/src/main/java/org/apache/blur/server/TableContext.java +++ b/blur-core/src/main/java/org/apache/blur/server/TableContext.java @@ -19,6 +19,7 @@ package org.apache.blur.server; import static org.apache.blur.utils.BlurConstants.BLUR_FIELDTYPE; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_BLURINDEX_CLASS; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_INDEX_DELETION_POLICY_MAXAGE; +import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_INDEX_QUEUE_READER_CLASS; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_INDEX_SIMILARITY; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_READ_INTERCEPTOR; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_TIME_BETWEEN_COMMITS; @@ -49,6 +50,7 @@ import org.apache.blur.manager.indexserver.BlurIndexWarmup; import org.apache.blur.manager.writer.BlurIndex; import org.apache.blur.manager.writer.BlurIndexCloser; import org.apache.blur.manager.writer.BlurIndexSimpleWriter; +import org.apache.blur.manager.writer.QueueReader; //import org.apache.blur.manager.writer.BlurNRTIndex; import org.apache.blur.manager.writer.SharedMergeScheduler; import org.apache.blur.thrift.generated.ScoreType; @@ -364,4 +366,35 @@ public class TableContext { return _readInterceptor; } + @SuppressWarnings("unchecked") + public QueueReader getQueueReader(BlurIndex blurIndex, ShardContext shardContext) throws IOException { + String className = _blurConfiguration.get(BLUR_SHARD_INDEX_QUEUE_READER_CLASS); + if (className == null || className.trim().isEmpty()) { + return null; + } + Class<? extends QueueReader> clazz; + try { + clazz = (Class<? extends QueueReader>) Class.forName(className); + } catch (ClassNotFoundException e) { + throw new IOException(e); + } + try { + Constructor<? extends QueueReader> constructor = clazz.getConstructor(new Class[] { BlurIndex.class, + ShardContext.class }); + return constructor.newInstance(blurIndex, shardContext); + } catch (NoSuchMethodException e) { + throw new IOException(e); + } catch (SecurityException e) { + throw new IOException(e); + } catch (InstantiationException e) { + throw new IOException(e); + } catch (IllegalAccessException e) { + throw new IOException(e); + } catch (IllegalArgumentException e) { + throw new IOException(e); + } catch (InvocationTargetException e) { + throw new IOException(e); + } + } + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/052c131e/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexSimpleWriterTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexSimpleWriterTest.java b/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexSimpleWriterTest.java index 618b24b..2ca0447 100644 --- a/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexSimpleWriterTest.java +++ b/blur-core/src/test/java/org/apache/blur/manager/writer/BlurIndexSimpleWriterTest.java @@ -63,7 +63,6 @@ public class BlurIndexSimpleWriterTest { private SharedMergeScheduler _mergeScheduler; private String uuid; - private BlurIndexRefresher _refresher; private BlurIndexCloser _closer; private DefaultBlurIndexWarmup _indexWarmup; @@ -78,7 +77,6 @@ public class BlurIndexSimpleWriterTest { _configuration = new Configuration(); _service = Executors.newThreadPool("test", 10); - _refresher = new BlurIndexRefresher(); _closer = new BlurIndexCloser(); _indexWarmup = new DefaultBlurIndexWarmup(1000000); } @@ -107,7 +105,6 @@ public class BlurIndexSimpleWriterTest { @After public void tearDown() throws IOException { - _refresher.close(); _writer.close(); _mergeScheduler.close(); _service.shutdownNow(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/052c131e/blur-core/src/test/java/org/apache/blur/manager/writer/QueueReaderBasicInMemory.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/writer/QueueReaderBasicInMemory.java b/blur-core/src/test/java/org/apache/blur/manager/writer/QueueReaderBasicInMemory.java new file mode 100644 index 0000000..9c9507a --- /dev/null +++ b/blur-core/src/test/java/org/apache/blur/manager/writer/QueueReaderBasicInMemory.java @@ -0,0 +1,49 @@ +/** + * 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.util.List; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; + +import org.apache.blur.server.ShardContext; +import org.apache.blur.thrift.generated.RowMutation; + +public class QueueReaderBasicInMemory extends QueueReader { + + static final BlockingQueue<RowMutation> _queue = new ArrayBlockingQueue<RowMutation>(1000); + + public QueueReaderBasicInMemory(BlurIndex index, ShardContext shardContext) { + super(index, shardContext); + } + + @Override + public void take(List<RowMutation> mutations, int max) { + _queue.drainTo(mutations, max); + } + + @Override + public void success() { + + } + + @Override + public void failure() { + + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/052c131e/blur-core/src/test/java/org/apache/blur/manager/writer/QueueReaderTest.java ---------------------------------------------------------------------- diff --git a/blur-core/src/test/java/org/apache/blur/manager/writer/QueueReaderTest.java b/blur-core/src/test/java/org/apache/blur/manager/writer/QueueReaderTest.java new file mode 100644 index 0000000..c58b1cd --- /dev/null +++ b/blur-core/src/test/java/org/apache/blur/manager/writer/QueueReaderTest.java @@ -0,0 +1,158 @@ +/** + * 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.File; +import java.io.IOException; +import java.util.Random; +import java.util.UUID; +import java.util.concurrent.ExecutorService; + +import org.apache.blur.concurrent.Executors; +import org.apache.blur.manager.indexserver.DefaultBlurIndexWarmup; +import org.apache.blur.server.ShardContext; +import org.apache.blur.server.TableContext; +import org.apache.blur.thrift.generated.Column; +import org.apache.blur.thrift.generated.Record; +import org.apache.blur.thrift.generated.RecordMutation; +import org.apache.blur.thrift.generated.RecordMutationType; +import org.apache.blur.thrift.generated.RowMutation; +import org.apache.blur.thrift.generated.RowMutationType; +import org.apache.blur.thrift.generated.TableDescriptor; +import org.apache.blur.utils.BlurConstants; +import org.apache.hadoop.conf.Configuration; +import org.apache.lucene.store.FSDirectory; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class QueueReaderTest { + + private static final int TOTAL_ROWS_FOR_TESTS = 10000; + private static final String TEST_TABLE = "test-table"; + private static final File TMPDIR = new File("./target/tmp"); + + private BlurIndexSimpleWriter _writer; + private Random random = new Random(); + private ExecutorService _service; + private File _base; + private Configuration _configuration; + + private SharedMergeScheduler _mergeScheduler; + private String uuid; + private BlurIndexCloser _closer; + private DefaultBlurIndexWarmup _indexWarmup; + + @Before + public void setup() throws IOException { + TableContext.clear(); + _base = new File(TMPDIR, "QueueReaderTest"); + rmr(_base); + _base.mkdirs(); + + _mergeScheduler = new SharedMergeScheduler(1); + + _configuration = new Configuration(); + _service = Executors.newThreadPool("test", 10); + _closer = new BlurIndexCloser(); + _indexWarmup = new DefaultBlurIndexWarmup(1000000); + } + + private void setupWriter(Configuration configuration) throws IOException { + TableDescriptor tableDescriptor = new TableDescriptor(); + tableDescriptor.setName(TEST_TABLE); + /* + * if reload is set to true...we create a new writer instance pointing to + * the same location as the old one..... so previous writer instances should + * be closed + */ + + uuid = UUID.randomUUID().toString(); + + tableDescriptor.setTableUri(new File(_base, "table-store-" + uuid).toURI().toString()); + tableDescriptor.putToTableProperties(BlurConstants.BLUR_SHARD_INDEX_QUEUE_READER_CLASS, + QueueReaderBasicInMemory.class.getName()); + TableContext tableContext = TableContext.create(tableDescriptor); + File path = new File(_base, "index_" + uuid); + path.mkdirs(); + FSDirectory directory = FSDirectory.open(path); + ShardContext shardContext = ShardContext.create(tableContext, "test-shard-" + uuid); + _writer = new BlurIndexSimpleWriter(shardContext, directory, _mergeScheduler, _service, _closer, _indexWarmup); + } + + @After + public void tearDown() throws IOException { + _writer.close(); + _mergeScheduler.close(); + _service.shutdownNow(); + rmr(_base); + } + + private void rmr(File file) { + if (!file.exists()) { + return; + } + if (file.isDirectory()) { + for (File f : file.listFiles()) { + rmr(f); + } + } + file.delete(); + } + + @Test + public void testQueueReader() throws IOException, InterruptedException { + System.out.println(_configuration.get(BlurConstants.BLUR_SHARD_INDEX_QUEUE_READER_CLASS)); + setupWriter(_configuration); + final String table = _writer.getShardContext().getTableContext().getTable(); + new Thread(new Runnable() { + @Override + public void run() { + for (int i = 0; i < TOTAL_ROWS_FOR_TESTS; i++) { + try { + QueueReaderBasicInMemory._queue.put(genRowMutation(table)); + } catch (InterruptedException e) { + return; + } + } + } + }).start(); + while (true) { + if (_writer.getIndexSearcher().getIndexReader().numDocs() == TOTAL_ROWS_FOR_TESTS) { + break; + } + Thread.sleep(100); + } + // YAY!!! it worked! + } + + private RowMutation genRowMutation(String table) { + RowMutation rowMutation = new RowMutation(); + rowMutation.setRowId(Long.toString(random.nextLong())); + rowMutation.setTable(table); + rowMutation.setRowMutationType(RowMutationType.REPLACE_ROW); + Record record = new Record(); + record.setFamily("testing"); + record.setRecordId(Long.toString(random.nextLong())); + for (int i = 0; i < 10; i++) { + record.addToColumns(new Column("col" + i, Long.toString(random.nextLong()))); + } + rowMutation.addToRecordMutations(new RecordMutation(RecordMutationType.REPLACE_ENTIRE_RECORD, record)); + return rowMutation; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/052c131e/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java b/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java index a6d7824..95052da 100644 --- a/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java +++ b/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java @@ -37,6 +37,10 @@ public class BlurConstants { public static final String SUPER = "super"; public static final String SEP = "."; + public static final String BLUR_SHARD_INDEX_QUEUE_READER_CLASS = "blur.shard.index.queue.reader.class"; + public static final String BLUR_SHARD_INDEX_QUEUE_READER_BACKOFF = "blur.shard.index.queue.reader.backoff"; + public static final String BLUR_SHARD_INDEX_QUEUE_READER_MAX = "blur.shard.index.queue.reader.max"; + public static final String FAST_DECOMPRESSION = "FAST_DECOMPRESSION"; public static final String FAST = "FAST"; public static final String HIGH_COMPRESSION = "HIGH_COMPRESSION";
