Repository: incubator-blur Updated Branches: refs/heads/apache-blur-0.2 cdc63a8be -> ef83c6021
let subclasses fully initialize before starting the daemon Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/ef83c602 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/ef83c602 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/ef83c602 Branch: refs/heads/apache-blur-0.2 Commit: ef83c602184e179259eb889622ba807ad4085470 Parents: cdc63a8 Author: twilliams <[email protected]> Authored: Wed Feb 26 10:06:32 2014 -0500 Committer: twilliams <[email protected]> Committed: Wed Feb 26 10:06:32 2014 -0500 ---------------------------------------------------------------------- .../apache/blur/manager/writer/QueueReader.java | 63 +++++++++++--------- .../org/apache/blur/server/TableContext.java | 6 +- 2 files changed, 39 insertions(+), 30 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ef83c602/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 index f44dea9..a911465 100644 --- 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 @@ -32,7 +32,7 @@ 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 { +public abstract class QueueReader implements Closeable { private static final Log LOG = LogFactory.getLog(QueueReader.class); @@ -52,38 +52,43 @@ public abstract class QueueReader implements Closeable, Runnable { 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 = new Thread(new Runnable() { + + @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(); + } + } + } + } + }); _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(); - } - } - } + public void listen() { + _daemon.start(); } @Override http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/ef83c602/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 8c1f012..53a912b 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 @@ -381,7 +381,11 @@ public class TableContext { try { Constructor<? extends QueueReader> constructor = clazz.getConstructor(new Class[] { BlurIndex.class, ShardContext.class }); - return constructor.newInstance(blurIndex, shardContext); + QueueReader reader = constructor.newInstance(blurIndex, shardContext); + + reader.listen(); + + return reader; } catch (NoSuchMethodException e) { throw new IOException(e); } catch (SecurityException e) {
