Fixed BLUR-179
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/a93ab7a6 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/a93ab7a6 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/a93ab7a6 Branch: refs/heads/0.2.0-newtypesystem Commit: a93ab7a68b782769fe23f35b799ca14986eca32f Parents: 9267331 Author: Aaron McCurry <[email protected]> Authored: Thu Aug 1 16:49:58 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Thu Aug 1 16:49:58 2013 -0400 ---------------------------------------------------------------------- .../blur/thrift/BlurControllerServer.java | 37 +++++++++++++------- 1 file changed, 25 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/a93ab7a6/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java b/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java index 173cff1..c888535 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/BlurControllerServer.java @@ -813,23 +813,36 @@ public class BlurControllerServer extends TableAdmin implements Iface { batches.put(node, list); } list.add(mutation); - } + List<Future<Void>> futures = new ArrayList<Future<Void>>(); + for (Entry<String, List<RowMutation>> entry : batches.entrySet()) { - String node = entry.getKey(); + final String node = entry.getKey(); final List<RowMutation> mutationsLst = entry.getValue(); + futures.add(_executor.submit(new Callable<Void>() { + @Override + public Void call() throws Exception { + return _client.execute(node, new BlurCommand<Void>() { + @Override + public Void call(Client client) throws BlurException, TException { + client.mutateBatch(mutationsLst); + return null; + } + }, _maxMutateRetries, _mutateDelay, _maxMutateDelay); + } + })); + } + + for (Future<Void> future : futures) { try { - _client.execute(node, new BlurCommand<Void>() { - @Override - public Void call(Client client) throws BlurException, TException { - client.mutateBatch(mutationsLst); - return null; - } - }, _maxMutateRetries, _mutateDelay, _maxMutateDelay); - } catch (Exception e) { - LOG.error("Unknown error during mutations of [{0}]", e, mutationsLst); - throw new BException("Unknown error during mutations of [{0}]", e, mutationsLst); + future.get(); + } catch (InterruptedException e) { + LOG.error("Unknown error during batch mutations", e); + throw new BException("Unknown error during batch mutations", e); + } catch (ExecutionException e) { + LOG.error("Unknown error during batch mutations", e.getCause()); + throw new BException("Unknown error during batch mutations", e.getCause()); } } }
