Trying to improve error handling and exception propagation.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/6707ae0e Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/6707ae0e Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/6707ae0e Branch: refs/heads/master Commit: 6707ae0ef198f41cfe3d2a96e3a764c61d65331d Parents: d084a16 Author: Aaron McCurry <[email protected]> Authored: Tue Jan 14 20:33:50 2014 -0500 Committer: Aaron McCurry <[email protected]> Committed: Wed Jan 15 17:02:35 2014 -0500 ---------------------------------------------------------------------- .../blur/thrift/BlurControllerServer.java | 363 +++++++++++-------- .../org/apache/blur/thrift/BlurShardServer.java | 145 +++++--- .../blur/thrift/ThriftBlurControllerServer.java | 1 + .../blur/thrift/ThriftBlurShardServer.java | 1 + .../java/org/apache/blur/utils/BlurUtil.java | 25 ++ 5 files changed, 332 insertions(+), 203 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6707ae0e/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 67a838b..f02490c 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 @@ -353,23 +353,23 @@ public class BlurControllerServer extends TableAdmin implements Iface { @Override public BlurResults query(final String table, final BlurQuery blurQuery) throws BlurException, TException { - checkTable(table); - Tracer trace = Trace.trace("query - setup", Trace.param("table", table), Trace.param("blurQuery", blurQuery)); - String cluster = _clusterStatus.getCluster(true, table); - _queryChecker.checkQuery(blurQuery); - checkSelectorFetchSize(blurQuery.getSelector()); - TableDescriptor tableDescriptor = _clusterStatus.getTableDescriptor(true, cluster, table); - int shardCount = tableDescriptor.getShardCount(); - if (blurQuery.getUuid() == null) { - blurQuery.setUuid(UUID.randomUUID().toString()); - } - BlurUtil.setStartTime(blurQuery); - trace.done(); + try { + checkTable(table); + Tracer trace = Trace.trace("query - setup", Trace.param("table", table), Trace.param("blurQuery", blurQuery)); + String cluster = _clusterStatus.getCluster(true, table); + _queryChecker.checkQuery(blurQuery); + checkSelectorFetchSize(blurQuery.getSelector()); + TableDescriptor tableDescriptor = _clusterStatus.getTableDescriptor(true, cluster, table); + int shardCount = tableDescriptor.getShardCount(); + if (blurQuery.getUuid() == null) { + blurQuery.setUuid(UUID.randomUUID().toString()); + } + BlurUtil.setStartTime(blurQuery); + trace.done(); - BlurUtil.setStartTime(blurQuery); + BlurUtil.setStartTime(blurQuery); - OUTER: for (int retries = 0; retries < _maxDefaultRetries; retries++) { - try { + OUTER: for (int retries = 0; retries < _maxDefaultRetries; retries++) { Tracer selectorTrace = Trace.trace("selector - setup", Trace.param("retries", retries)); final AtomicLongArray facetCounts = BlurUtil.getAtomicLongArraySameLengthAsList(blurQuery.facets); Selector selector = blurQuery.getSelector(); @@ -431,15 +431,15 @@ public class BlurControllerServer extends TableAdmin implements Iface { hitsIterable.close(); } } - } catch (Exception e) { - if (e instanceof BlurException) { - throw (BlurException) e; - } - LOG.error("Unknown error during search of [table={0},blurQuery={1}]", e, table, blurQuery); - throw new BException("Unknown error during search of [table={0},blurQuery={1}]", e, table, blurQuery); } + throw new BException("Query could not be completed."); + } catch (Exception e) { + LOG.error("Unknown error during search of [table={0},blurQuery={1}]", e, table, blurQuery); + if (e instanceof BlurException) { + throw (BlurException) e; + } + throw new BException("Unknown error during search of [table={0},blurQuery={1}]", e, table, blurQuery); } - throw new BException("Query could not be completed."); } public BlurResults convertToBlurResults(BlurResultIterable hitsIterable, BlurQuery query, @@ -573,11 +573,11 @@ public class BlurControllerServer extends TableAdmin implements Iface { @Override public FetchResult fetchRow(final String table, final Selector selector) throws BlurException, TException { - checkTable(table); - checkSelectorFetchSize(selector); - IndexManager.validSelector(selector); String clientHostnamePort = null; try { + checkTable(table); + checkSelectorFetchSize(selector); + IndexManager.validSelector(selector); clientHostnamePort = getNode(table, selector); return _client.execute(clientHostnamePort, new BlurCommand<FetchResult>() { @Override @@ -588,6 +588,9 @@ public class BlurControllerServer extends TableAdmin implements Iface { } catch (Exception e) { LOG.error("Unknown error during fetch of row from table [{0}] selector [{1}] node [{2}]", e, table, selector, clientHostnamePort); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException("Unknown error during fetch of row from table [{0}] selector [{1}] node [{2}]", e, table, selector, clientHostnamePort); } @@ -595,76 +598,85 @@ public class BlurControllerServer extends TableAdmin implements Iface { @Override public List<FetchResult> fetchRowBatch(final String table, List<Selector> selectors) throws BlurException, TException { - checkTable(table); - Map<String, List<Selector>> selectorBatches = new HashMap<String, List<Selector>>(); - final Map<String, List<Integer>> selectorBatchesIndexes = new HashMap<String, List<Integer>>(); - int i = 0; - for (Selector selector : selectors) { - checkSelectorFetchSize(selector); - IndexManager.validSelector(selector); - String clientHostnamePort = getNode(table, selector); - List<Selector> list = selectorBatches.get(clientHostnamePort); - List<Integer> indexes = selectorBatchesIndexes.get(clientHostnamePort); - if (list == null) { - if (indexes != null) { - throw new BlurException("This should never happen,", null, ErrorType.UNKNOWN); + try { + checkTable(table); + Map<String, List<Selector>> selectorBatches = new HashMap<String, List<Selector>>(); + final Map<String, List<Integer>> selectorBatchesIndexes = new HashMap<String, List<Integer>>(); + int i = 0; + for (Selector selector : selectors) { + checkSelectorFetchSize(selector); + IndexManager.validSelector(selector); + String clientHostnamePort = getNode(table, selector); + List<Selector> list = selectorBatches.get(clientHostnamePort); + List<Integer> indexes = selectorBatchesIndexes.get(clientHostnamePort); + if (list == null) { + if (indexes != null) { + throw new BlurException("This should never happen,", null, ErrorType.UNKNOWN); + } + list = new ArrayList<Selector>(); + indexes = new ArrayList<Integer>(); + selectorBatches.put(clientHostnamePort, list); + selectorBatchesIndexes.put(clientHostnamePort, indexes); } - list = new ArrayList<Selector>(); - indexes = new ArrayList<Integer>(); - selectorBatches.put(clientHostnamePort, list); - selectorBatchesIndexes.put(clientHostnamePort, indexes); + list.add(selector); + indexes.add(i); + i++; } - list.add(selector); - indexes.add(i); - i++; - } - List<Future<Boolean>> futures = new ArrayList<Future<Boolean>>(); - final AtomicReferenceArray<FetchResult> fetchResults = new AtomicReferenceArray<FetchResult>(new FetchResult[i]); - for (Entry<String, List<Selector>> batch : selectorBatches.entrySet()) { - final String clientHostnamePort = batch.getKey(); - final List<Selector> list = batch.getValue(); - futures.add(_executor.submit(new Callable<Boolean>() { - @Override - public Boolean call() throws Exception { - List<FetchResult> fetchResultList = _client.execute(clientHostnamePort, new BlurCommand<List<FetchResult>>() { - @Override - public List<FetchResult> call(Client client) throws BlurException, TException { - return client.fetchRowBatch(table, list); + List<Future<Boolean>> futures = new ArrayList<Future<Boolean>>(); + final AtomicReferenceArray<FetchResult> fetchResults = new AtomicReferenceArray<FetchResult>(new FetchResult[i]); + for (Entry<String, List<Selector>> batch : selectorBatches.entrySet()) { + final String clientHostnamePort = batch.getKey(); + final List<Selector> list = batch.getValue(); + futures.add(_executor.submit(new Callable<Boolean>() { + @Override + public Boolean call() throws Exception { + List<FetchResult> fetchResultList = _client.execute(clientHostnamePort, + new BlurCommand<List<FetchResult>>() { + @Override + public List<FetchResult> call(Client client) throws BlurException, TException { + return client.fetchRowBatch(table, list); + } + }, _maxFetchRetries, _fetchDelay, _maxFetchDelay); + List<Integer> indexes = selectorBatchesIndexes.get(clientHostnamePort); + for (int i = 0; i < fetchResultList.size(); i++) { + int index = indexes.get(i); + fetchResults.set(index, fetchResultList.get(i)); } - }, _maxFetchRetries, _fetchDelay, _maxFetchDelay); - List<Integer> indexes = selectorBatchesIndexes.get(clientHostnamePort); - for (int i = 0; i < fetchResultList.size(); i++) { - int index = indexes.get(i); - fetchResults.set(index, fetchResultList.get(i)); + return Boolean.TRUE; } - return Boolean.TRUE; - } - })); - } + })); + } - for (Future<Boolean> future : futures) { - try { - future.get(); - } catch (InterruptedException e) { - throw new BException("Unknown error during fetching of batch", e); - } catch (ExecutionException e) { - throw new BException("Unknown error during fetching of batch", e.getCause()); + for (Future<Boolean> future : futures) { + try { + future.get(); + } catch (InterruptedException e) { + throw new BException("Unknown error during fetching of batch", e); + } catch (ExecutionException e) { + throw new BException("Unknown error during fetching of batch", e.getCause()); + } } - } - List<FetchResult> batchResult = new ArrayList<FetchResult>(); - for (int c = 0; c < fetchResults.length(); c++) { - FetchResult fetchResult = fetchResults.get(c); - batchResult.add(fetchResult); + List<FetchResult> batchResult = new ArrayList<FetchResult>(); + for (int c = 0; c < fetchResults.length(); c++) { + FetchResult fetchResult = fetchResults.get(c); + batchResult.add(fetchResult); + } + return batchResult; + } catch (Exception e) { + LOG.error("Unknown error during fetch a batch of rows from table [{0}]", e, table); + if (e instanceof BlurException) { + throw (BlurException) e; + } + throw new BException("Unknown error during fetch a batch of rows from table [{0}]", e, table); } - return batchResult; } @Override public void cancelQuery(final String table, final String uuid) throws BlurException, TException { - checkTable(table); try { + checkTable(table); scatter(getCluster(table), new BlurCommand<Void>() { @Override public Void call(Client client) throws BlurException, TException { @@ -674,14 +686,17 @@ public class BlurControllerServer extends TableAdmin implements Iface { }); } catch (Exception e) { LOG.error("Unknown error while trying to cancel search table [{0}] uuid [{1}]", e, table, uuid); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException("Unknown error while trying to cancel search table [{0}] uuid [{1}]", e, table, uuid); } } @Override public List<String> queryStatusIdList(final String table) throws BlurException, TException { - checkTable(table); try { + checkTable(table); return scatterGather(getCluster(table), new BlurCommand<List<String>>() { @Override public List<String> call(Client client) throws BlurException, TException { @@ -701,14 +716,17 @@ public class BlurControllerServer extends TableAdmin implements Iface { }); } catch (Exception e) { LOG.error("Unknown error while trying to get query status ids for table [{0}]", e, table); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException("Unknown error while trying to get query status ids for table [{0}]", e, table); } } @Override public BlurQueryStatus queryStatusById(final String table, final String uuid) throws BlurException, TException { - checkTable(table); try { + checkTable(table); return scatterGather(getCluster(table), new BlurCommand<BlurQueryStatus>() { @Override public BlurQueryStatus call(Client client) throws BlurException, TException { @@ -717,14 +735,17 @@ public class BlurControllerServer extends TableAdmin implements Iface { }, new MergerQueryStatusSingle(_defaultParallelCallTimeout)); } catch (Exception e) { LOG.error("Unknown error while trying to get query status [{0}]", e, table, uuid); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException("Unknown error while trying to get query status [{0}]", e, table, uuid); } } @Override public TableStats tableStats(final String table) throws BlurException, TException { - checkTable(table); try { + checkTable(table); return scatterGather(getCluster(table), new BlurCommand<TableStats>() { @Override public TableStats call(Client client) throws BlurException, TException { @@ -733,19 +754,30 @@ public class BlurControllerServer extends TableAdmin implements Iface { }, new MergerTableStats(_defaultParallelCallTimeout)); } catch (Exception e) { LOG.error("Unknown error while trying to get table stats [{0}]", e, table); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException("Unknown error while trying to get table stats [{0}]", e, table); } } @Override public Map<String, String> shardServerLayout(String table) throws BlurException, TException { - checkTable(table); - Map<String, Map<String, String>> layout = _shardServerLayout.get(); - Map<String, String> tableLayout = layout.get(table); - if (tableLayout == null) { - return new HashMap<String, String>(); + try { + checkTable(table); + Map<String, Map<String, String>> layout = _shardServerLayout.get(); + Map<String, String> tableLayout = layout.get(table); + if (tableLayout == null) { + return new HashMap<String, String>(); + } + return tableLayout; + } catch (Exception e) { + LOG.error("Unknown error while trying to get shard server layout [{0}]", e, table); + if (e instanceof BlurException) { + throw (BlurException) e; + } + throw new BException("Unknown error while trying to get shard server layout [{0}]", e, table); } - return tableLayout; } @Override @@ -785,6 +817,9 @@ public class BlurControllerServer extends TableAdmin implements Iface { }); } catch (Exception e) { LOG.error("Unknown error while trying to get shard server layout [{0}]", e, table); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException("Unknown error while trying to get shard server layout [{0}]", e, table); } } @@ -792,8 +827,8 @@ public class BlurControllerServer extends TableAdmin implements Iface { @Override public long recordFrequency(final String table, final String columnFamily, final String columnName, final String value) throws BlurException, TException { - checkTable(table); try { + checkTable(table); return scatterGather(getCluster(table), new BlurCommand<Long>() { @Override public Long call(Client client) throws BlurException, TException { @@ -815,6 +850,9 @@ public class BlurControllerServer extends TableAdmin implements Iface { } catch (Exception e) { LOG.error("Unknown error while trying to get record frequency [{0}/{1}/{2}/{3}]", e, table, columnFamily, columnName, value); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException("Unknown error while trying to get record frequency [{0}/{1}/{2}/{3}]", e, table, columnFamily, columnName, value); } @@ -823,8 +861,8 @@ public class BlurControllerServer extends TableAdmin implements Iface { @Override public List<String> terms(final String table, final String columnFamily, final String columnName, final String startWith, final short size) throws BlurException, TException { - checkTable(table); try { + checkTable(table); return scatterGather(getCluster(table), new BlurCommand<List<String>>() { @Override public List<String> call(Client client) throws BlurException, TException { @@ -846,6 +884,9 @@ public class BlurControllerServer extends TableAdmin implements Iface { LOG.error( "Unknown error while trying to terms table [{0}] columnFamily [{1}] columnName [{2}] startWith [{3}] size [{4}]", e, table, columnFamily, columnName, startWith, size); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException( "Unknown error while trying to terms table [{0}] columnFamily [{1}] columnName [{2}] startWith [{3}] size [{4}]", e, table, columnFamily, columnName, startWith, size); @@ -915,9 +956,9 @@ public class BlurControllerServer extends TableAdmin implements Iface { @Override public void mutate(final RowMutation mutation) throws BlurException, TException { - checkTable(mutation.table); - checkForUpdates(mutation.table); try { + checkTable(mutation.table); + checkForUpdates(mutation.table); MutationHelper.validateMutation(mutation); String table = mutation.getTable(); @@ -938,6 +979,9 @@ public class BlurControllerServer extends TableAdmin implements Iface { }, _maxMutateRetries, _mutateDelay, _maxMutateDelay); } catch (Exception e) { LOG.error("Unknown error during mutation of [{0}]", e, mutation); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException("Unknown error during mutation of [{0}]", e, mutation); } } @@ -954,69 +998,77 @@ public class BlurControllerServer extends TableAdmin implements Iface { @Override public void mutateBatch(List<RowMutation> mutations) throws BlurException, TException { - for (RowMutation mutation : mutations) { - MutationHelper.validateMutation(mutation); - } - Map<String, List<RowMutation>> batches = new HashMap<String, List<RowMutation>>(); - for (RowMutation mutation : mutations) { - checkTable(mutation.table); - checkForUpdates(mutation.table); - - MutationHelper.validateMutation(mutation); - String table = mutation.getTable(); - - int numberOfShards = getShardCount(table); - Map<String, String> tableLayout = _shardServerLayout.get().get(table); - if (tableLayout == null || tableLayout.size() != numberOfShards) { - throw new BException("Cannot update data while shard is missing"); + try { + for (RowMutation mutation : mutations) { + MutationHelper.validateMutation(mutation); } + Map<String, List<RowMutation>> batches = new HashMap<String, List<RowMutation>>(); + for (RowMutation mutation : mutations) { + checkTable(mutation.table); + checkForUpdates(mutation.table); + + MutationHelper.validateMutation(mutation); + String table = mutation.getTable(); + + int numberOfShards = getShardCount(table); + Map<String, String> tableLayout = _shardServerLayout.get().get(table); + if (tableLayout == null || tableLayout.size() != numberOfShards) { + throw new BException("Cannot update data while shard is missing"); + } - String shardName = MutationHelper.getShardName(table, mutation.rowId, numberOfShards, _blurPartitioner); - String node = tableLayout.get(shardName); - List<RowMutation> list = batches.get(node); - if (list == null) { - list = new ArrayList<RowMutation>(); - batches.put(node, list); + String shardName = MutationHelper.getShardName(table, mutation.rowId, numberOfShards, _blurPartitioner); + String node = tableLayout.get(shardName); + List<RowMutation> list = batches.get(node); + if (list == null) { + list = new ArrayList<RowMutation>(); + batches.put(node, list); + } + list.add(mutation); } - list.add(mutation); - } - List<Future<Void>> futures = new ArrayList<Future<Void>>(); + List<Future<Void>> futures = new ArrayList<Future<Void>>(); - for (Entry<String, List<RowMutation>> entry : batches.entrySet()) { - 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 (Entry<String, List<RowMutation>> entry : batches.entrySet()) { + 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 { - 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()); + for (Future<Void> future : futures) { + try { + 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()); + } + } + } catch (Exception e) { + LOG.error("Unknown error during batch mutation.", e); + if (e instanceof BlurException) { + throw (BlurException) e; } + throw new BException("Unknown error during mutation.", e); } } @Override public void createSnapshot(final String table, final String name) throws BlurException, TException { - checkTable(table); try { + checkTable(table); scatter(getCluster(table), new BlurCommand<Void>() { @Override public Void call(Client client) throws BlurException, TException { @@ -1025,18 +1077,18 @@ public class BlurControllerServer extends TableAdmin implements Iface { } }); } catch (Exception e) { + LOG.error("Unknown error while trying to create a snapshot [{0}] of table [{1}]", e, name, table); if (e instanceof BlurException) { throw (BlurException) e; } - LOG.error("Unknown error while trying to create a snapshot [{0}] of table [{1}]", e, name, table); throw new BException("Unknown error while trying to create a snapshot [{0}] of table [{1}]", e, name, table); } } @Override public void removeSnapshot(final String table, final String name) throws BlurException, TException { - checkTable(table); try { + checkTable(table); scatter(getCluster(table), new BlurCommand<Void>() { @Override public Void call(Client client) throws BlurException, TException { @@ -1045,18 +1097,18 @@ public class BlurControllerServer extends TableAdmin implements Iface { } }); } catch (Exception e) { + LOG.error("Unknown error while trying to remove a snapshot [{0}] of table [{1}]", e, name, table); if (e instanceof BlurException) { throw (BlurException) e; } - LOG.error("Unknown error while trying to remove a snapshot [{0}] of table [{1}]", e, name, table); throw new BException("Unknown error while trying to remove a snapshot [{0}] of table [{1}]", e, name, table); } } @Override public Map<String, List<String>> listSnapshots(final String table) throws BlurException, TException { - checkTable(table); try { + checkTable(table); return scatterGather(getCluster(table), new BlurCommand<Map<String, List<String>>>() { @Override public Map<String, List<String>> call(Client client) throws BlurException, TException { @@ -1084,10 +1136,10 @@ public class BlurControllerServer extends TableAdmin implements Iface { } }); } catch (Exception e) { + LOG.error("Unknown error while trying to get the list of snapshots for table [{0}]", e, table); if (e instanceof BlurException) { throw (BlurException) e; } - LOG.error("Unknown error while trying to get the list of snapshots for table [{0}]", e, table); throw new BException("Unknown error while trying to get the list of snapshots for table [{0}]", e, table); } } @@ -1160,6 +1212,7 @@ public class BlurControllerServer extends TableAdmin implements Iface { public void optimize(final String table, final int numberOfSegmentsPerShard) throws BlurException, TException { checkTable(table); try { + checkTable(table); scatter(getCluster(table), new BlurCommand<Void>() { @Override public Void call(Client client) throws BlurException, TException { @@ -1170,6 +1223,9 @@ public class BlurControllerServer extends TableAdmin implements Iface { } catch (Exception e) { LOG.error("Unknown error while trying to optimize [table={0},numberOfSegmentsPerShard={1}]", e, table, numberOfSegmentsPerShard); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException("Unknown error while trying to optimize [table={0},numberOfSegmentsPerShard={1}]", e, table, numberOfSegmentsPerShard); } @@ -1177,10 +1233,10 @@ public class BlurControllerServer extends TableAdmin implements Iface { @Override public String parseQuery(final String table, final Query simpleQuery) throws BlurException, TException { - checkTable(table); - String cluster = getCluster(table); - List<String> onlineShardServers = _clusterStatus.getOnlineShardServers(true, cluster); try { + checkTable(table); + String cluster = getCluster(table); + List<String> onlineShardServers = _clusterStatus.getOnlineShardServers(true, cluster); return BlurClientManager.execute(getConnections(onlineShardServers), new BlurCommand<String>() { @Override public String call(Client client) throws BlurException, TException { @@ -1189,6 +1245,9 @@ public class BlurControllerServer extends TableAdmin implements Iface { }); } catch (Exception e) { LOG.error("Unknown error while trying to parse query [table={0},simpleQuery={1}]", e, table, simpleQuery); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException("Unknown error while trying to parse query [table={0},simpleQuery={1}]", e, table, simpleQuery); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6707ae0e/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 b750cba..ecd01c0 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 @@ -20,7 +20,6 @@ import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_CACHE_MAX_QUERYCACH import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_CACHE_MAX_TIMETOLIVE; import static org.apache.blur.utils.BlurConstants.BLUR_SHARD_DATA_FETCH_THREAD_COUNT; -import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -92,10 +91,10 @@ public class BlurShardServer extends TableAdmin implements Iface { @Override public BlurResults query(String table, BlurQuery blurQuery) throws BlurException, TException { - checkTable(_cluster, table); - resetSearchers(); - _queryChecker.checkQuery(blurQuery); try { + checkTable(_cluster, table); + resetSearchers(); + _queryChecker.checkQuery(blurQuery); checkSelectorFetchSize(blurQuery.getSelector()); BlurQuery original = new BlurQuery(blurQuery); Selector selector = original.getSelector(); @@ -142,8 +141,11 @@ public class BlurShardServer extends TableAdmin implements Iface { hitsIterable.close(); } } - } catch (IOException e) { + } catch (Exception e) { LOG.error("Unknown error during search of [table={0},searchQuery={1}]", e, table, blurQuery); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException(e.getMessage(), e); } } @@ -151,49 +153,63 @@ public class BlurShardServer extends TableAdmin implements Iface { @Override public String parseQuery(String table, Query simpleQuery) throws BlurException, TException { try { + checkTable(_cluster, table); + resetSearchers(); return _indexManager.parseQuery(table, simpleQuery); - } catch (Throwable e) { + } catch (Exception e) { LOG.error("Unknown error during parsing of [table={0},simpleQuery={1}]", e, table, simpleQuery); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException(e.getMessage(), e); } } @Override public FetchResult fetchRow(String table, Selector selector) throws BlurException, TException { - checkTable(_cluster, table); - checkSelectorFetchSize(selector); try { + checkTable(_cluster, table); + checkSelectorFetchSize(selector); FetchResult fetchResult = new FetchResult(); _indexManager.fetchRow(table, selector, fetchResult); return fetchResult; } catch (Exception e) { LOG.error("Unknown error while trying to get fetch row [table={0},selector={1}]", e, table, selector); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException(e.getMessage(), e); } } @Override public List<FetchResult> fetchRowBatch(String table, List<Selector> selectors) throws BlurException, TException { - checkTable(_cluster, table); - for (Selector selector : selectors) { - checkSelectorFetchSize(selector); - } try { + checkTable(_cluster, table); + for (Selector selector : selectors) { + checkSelectorFetchSize(selector); + } return _indexManager.fetchRowBatch(table, selectors); } catch (Exception e) { LOG.error("Unknown error while trying to get fetch row [table={0},selector={1}]", e, table, selectors); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException(e.getMessage(), e); } } @Override public void cancelQuery(String table, String uuid) throws BlurException, TException { - checkTable(_cluster, table); - resetSearchers(); try { + checkTable(_cluster, table); + resetSearchers(); _indexManager.cancelQuery(table, uuid); } catch (Exception e) { LOG.error("Unknown error while trying to cancel search [uuid={0}]", e, uuid); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException(e.getMessage(), e); } } @@ -204,9 +220,9 @@ public class BlurShardServer extends TableAdmin implements Iface { @Override public TableStats tableStats(String table) throws BlurException, TException { - checkTable(_cluster, table); - resetSearchers(); try { + checkTable(_cluster, table); + resetSearchers(); TableStats tableStats = new TableStats(); tableStats.tableName = table; tableStats.recordCount = _indexServer.getRecordCount(table); @@ -215,6 +231,9 @@ public class BlurShardServer extends TableAdmin implements Iface { return tableStats; } catch (Exception e) { LOG.error("Unknown error while trying to get table stats [table={0}]", e, table); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException(e.getMessage(), e); } } @@ -229,9 +248,9 @@ public class BlurShardServer extends TableAdmin implements Iface { @Override public Map<String, String> shardServerLayout(String table) throws BlurException, TException { - checkTable(_cluster, table); - resetSearchers(); try { + checkTable(_cluster, table); + resetSearchers(); Map<String, BlurIndex> blurIndexes = _indexServer.getIndexes(table); Map<String, String> result = new TreeMap<String, String>(); String nodeName = _indexServer.getNodeName(); @@ -250,8 +269,8 @@ public class BlurShardServer extends TableAdmin implements Iface { @Override public Map<String, Map<String, ShardState>> shardServerLayoutState(String table) throws BlurException, TException { - resetSearchers(); try { + resetSearchers(); Map<String, Map<String, ShardState>> result = new TreeMap<String, Map<String, ShardState>>(); String nodeName = _indexServer.getNodeName(); Map<String, ShardState> stateMap = _indexServer.getShardState(table); @@ -277,70 +296,87 @@ public class BlurShardServer extends TableAdmin implements Iface { @Override public long recordFrequency(String table, String columnFamily, String columnName, String value) throws BlurException, TException { - checkTable(_cluster, table); - resetSearchers(); try { + checkTable(_cluster, table); + resetSearchers(); return _indexManager.recordFrequency(table, columnFamily, columnName, value); } catch (Exception e) { LOG.error( "Unknown error while trying to get record frequency for [table={0},columnFamily={1},columnName={2},value={3}]", e, table, columnFamily, columnName, value); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException(e.getMessage(), e); } } @Override public Schema schema(String table) throws BlurException, TException { - resetSearchers(); - return super.schema(table); + try { + resetSearchers(); + return super.schema(table); + } catch (Exception e) { + LOG.error("Unknown error while trying to get schema for [table={0}]", e, table); + if (e instanceof BlurException) { + throw (BlurException) e; + } + throw new BException(e.getMessage(), e); + } } @Override public List<String> terms(String table, String columnFamily, String columnName, String startWith, short size) throws BlurException, TException { - checkTable(_cluster, table); - resetSearchers(); try { + checkTable(_cluster, table); + resetSearchers(); return _indexManager.terms(table, columnFamily, columnName, startWith, size); } catch (Exception e) { LOG.error( "Unknown error while trying to get terms list for [table={0},columnFamily={1},columnName={2},startWith={3},size={4}]", e, table, columnFamily, columnName, startWith, size); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException(e.getMessage(), e); } } @Override public void mutate(RowMutation mutation) throws BlurException, TException { - checkTable(_cluster, mutation.table); - checkForUpdates(_cluster, mutation.table); - resetSearchers(); - MutationHelper.validateMutation(mutation); try { + checkTable(_cluster, mutation.table); + checkForUpdates(_cluster, mutation.table); + resetSearchers(); + MutationHelper.validateMutation(mutation); _indexManager.mutate(mutation); } catch (Exception e) { LOG.error("Unknown error during processing of [mutation={0}]", e, mutation); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException(e.getMessage(), e); } } @Override public void mutateBatch(List<RowMutation> mutations) throws BlurException, TException { - resetSearchers(); - long s = System.nanoTime(); - for (RowMutation mutation : mutations) { - checkTable(_cluster, mutation.table); - checkForUpdates(_cluster, mutation.table); - MutationHelper.validateMutation(mutation); - } try { + resetSearchers(); + for (RowMutation mutation : mutations) { + checkTable(_cluster, mutation.table); + checkForUpdates(_cluster, mutation.table); + MutationHelper.validateMutation(mutation); + } _indexManager.mutate(mutations); } catch (Exception e) { LOG.error("Unknown error during processing of [mutations={0}]", e, mutations); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException(e.getMessage(), e); } - long e = System.nanoTime(); - LOG.debug("mutateBatch took [" + (e - s) / 1000000.0 + " ms] to complete"); } public long getMaxTimeToLive() { @@ -373,22 +409,25 @@ public class BlurShardServer extends TableAdmin implements Iface { @Override public BlurQueryStatus queryStatusById(String table, String uuid) throws BlurException, TException { - checkTable(_cluster, table); - resetSearchers(); - BlurQueryStatus blurQueryStatus; try { + checkTable(_cluster, table); + resetSearchers(); + BlurQueryStatus blurQueryStatus; blurQueryStatus = _indexManager.queryStatus(table, uuid); + if (blurQueryStatus == null) { + blurQueryStatus = new BlurQueryStatus(); + blurQueryStatus.status = Status.NOT_FOUND; + } else { + blurQueryStatus.status = Status.FOUND; + } + return blurQueryStatus; } catch (Exception e) { LOG.error("Unknown error while trying to get current query status [table={0},uuid={1}]", e, table, uuid); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException(e.getMessage(), e); } - if (blurQueryStatus == null) { - blurQueryStatus = new BlurQueryStatus(); - blurQueryStatus.status = Status.NOT_FOUND; - } else { - blurQueryStatus.status = Status.FOUND; - } - return blurQueryStatus; } @Override @@ -405,13 +444,16 @@ public class BlurShardServer extends TableAdmin implements Iface { @Override public void optimize(String table, int numberOfSegmentsPerShard) throws BlurException, TException { - checkTable(_cluster, table); - resetSearchers(); try { + checkTable(_cluster, table); + resetSearchers(); _indexManager.optimize(table, numberOfSegmentsPerShard); } catch (Exception e) { LOG.error("Unknown error while trying to optimize [table={0},numberOfSegmentsPerShard={1}]", e, table, numberOfSegmentsPerShard); + if (e instanceof BlurException) { + throw (BlurException) e; + } throw new BException(e.getMessage(), e); } } @@ -419,6 +461,7 @@ public class BlurShardServer extends TableAdmin implements Iface { @Override public void createSnapshot(final String table, final String name) throws BlurException, TException { try { + checkTable(_cluster, table); Map<String, BlurIndex> indexes = _indexServer.getIndexes(table); for (Entry<String, BlurIndex> entry : indexes.entrySet()) { BlurIndex index = entry.getValue(); @@ -452,8 +495,8 @@ public class BlurShardServer extends TableAdmin implements Iface { @Override public Map<String, List<String>> listSnapshots(final String table) throws BlurException, TException { - Map<String, List<String>> snapshots = new HashMap<String, List<String>>(); try { + Map<String, List<String>> snapshots = new HashMap<String, List<String>>(); Map<String, BlurIndex> indexes = _indexServer.getIndexes(table); for (Entry<String, BlurIndex> entry : indexes.entrySet()) { BlurIndex index = entry.getValue(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6707ae0e/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java index b571c00..c0a8175 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java @@ -153,6 +153,7 @@ public class ThriftBlurControllerServer extends ThriftServer { iface = BlurUtil.recordMethodCallsAndAverageTimes(iface, Iface.class, true); iface = BlurUtil.runWithUser(iface, true); iface = BlurUtil.runTrace(iface, true); + iface = BlurUtil.lastChanceErrorHandling(iface, Iface.class); int threadCount = configuration.getInt(BLUR_CONTROLLER_SERVER_THRIFT_THREAD_COUNT, 32); ControllerServerEventHandler eventHandler = new ControllerServerEventHandler(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6707ae0e/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java index 44c3a0e..d6f31ef 100644 --- a/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java +++ b/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java @@ -246,6 +246,7 @@ public class ThriftBlurShardServer extends ThriftServer { iface = BlurUtil.recordMethodCallsAndAverageTimes(iface, Iface.class, false); iface = BlurUtil.runWithUser(iface, false); iface = BlurUtil.runTrace(iface, false); + iface = BlurUtil.lastChanceErrorHandling(iface, Iface.class); if (httpServer != null) { WebAppContext context = httpServer.getContext(); context.addServlet(new ServletHolder(new TServlet(new Blur.Processor<Blur.Iface>(iface), http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/6707ae0e/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java b/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java index da1b0dd..3f14ddc 100644 --- a/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java +++ b/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java @@ -78,6 +78,7 @@ import org.apache.blur.thirdparty.thrift_0_9_0.TBase; import org.apache.blur.thirdparty.thrift_0_9_0.TException; import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TJSONProtocol; import org.apache.blur.thirdparty.thrift_0_9_0.transport.TMemoryBuffer; +import org.apache.blur.thrift.BException; import org.apache.blur.thrift.UserConverter; import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.thrift.generated.BlurException; @@ -162,6 +163,30 @@ public class BlurUtil { } @SuppressWarnings("unchecked") + public static <T extends Iface> T lastChanceErrorHandling(final T t, Class<T> clazz) { + InvocationHandler handler = new InvocationHandler() { + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + try { + return method.invoke(t, args); + } catch (InvocationTargetException e) { + Throwable targetException = e.getTargetException(); + if (targetException instanceof BlurException) { + throw targetException; + } else if (targetException instanceof TException) { + throw targetException; + } else { + throw new BException( + "Unknown error during call on method [{0}], this means that the method is handling exceptions correctly.", + targetException, method.getName()); + } + } + } + }; + return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class[] { clazz }, handler); + } + + @SuppressWarnings("unchecked") public static <T extends Iface> T recordMethodCallsAndAverageTimes(final T t, Class<T> clazz, final boolean controller) { final Map<String, Histogram> histogramMap = new ConcurrentHashMap<String, Histogram>(); Method[] declaredMethods = Iface.class.getDeclaredMethods();
