Repository: incubator-blur Updated Branches: refs/heads/master c0bb9f9fd -> da069f03d
More platform updates. Small API changes. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/da069f03 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/da069f03 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/da069f03 Branch: refs/heads/master Commit: da069f03d5770f1039770c6bf8d29eaa133e34f2 Parents: c0bb9f9 Author: Aaron McCurry <[email protected]> Authored: Thu Aug 28 15:27:00 2014 -0400 Committer: Aaron McCurry <[email protected]> Committed: Thu Aug 28 15:27:00 2014 -0400 ---------------------------------------------------------------------- .../manager/command/BaseCommandManager.java | 4 ++ .../command/ControllerClusterContext.java | 66 ++++++++++++++++++++ .../command/ControllerCommandManager.java | 60 ++++++++++++++---- .../blur/manager/command/IndexWriteCommand.java | 3 + .../apache/blur/manager/command/Response.java | 19 ++++-- .../manager/command/ShardCommandManager.java | 6 +- .../blur/thrift/BlurControllerServer.java | 11 +++- 7 files changed, 145 insertions(+), 24 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/da069f03/blur-core/src/main/java/org/apache/blur/manager/command/BaseCommandManager.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/command/BaseCommandManager.java b/blur-core/src/main/java/org/apache/blur/manager/command/BaseCommandManager.java index 4bd2626..c2183b6 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/command/BaseCommandManager.java +++ b/blur-core/src/main/java/org/apache/blur/manager/command/BaseCommandManager.java @@ -54,4 +54,8 @@ public class BaseCommandManager implements Closeable { throw new IOException(e); } } + + protected BaseCommand getCommandObject(String commandName) { + return _command.get(commandName); + } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/da069f03/blur-core/src/main/java/org/apache/blur/manager/command/ControllerClusterContext.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/command/ControllerClusterContext.java b/blur-core/src/main/java/org/apache/blur/manager/command/ControllerClusterContext.java new file mode 100644 index 0000000..de17e44 --- /dev/null +++ b/blur-core/src/main/java/org/apache/blur/manager/command/ControllerClusterContext.java @@ -0,0 +1,66 @@ +package org.apache.blur.manager.command; + +import java.io.Closeable; +import java.io.IOException; +import java.util.Map; + +import org.apache.blur.server.TableContext; + +/** + * 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. + */ + +public class ControllerClusterContext extends ClusterContext implements Closeable { + + private final Args _args; + private final TableContext _tableContext; + + public ControllerClusterContext(TableContext tableContext, Args args) { + _tableContext = tableContext; + _args = args; + } + + @Override + public Args getArgs() { + return _args; + } + + @Override + public TableContext getTableContext() { + return _tableContext; + } + + @Override + public <T> Map<Shard, T> readIndexes(Args args, Class<? extends IndexReadCommand<T>> clazz) { + throw new RuntimeException("Not Implemented"); + } + + @Override + public <T> Map<Server, T> readServers(Args args, Class<? extends IndexReadCombiningCommand<?, T>> clazz) { + throw new RuntimeException("Not Implemented"); + } + + @Override + public <T> T writeIndex(Args args, Class<? extends IndexWriteCommand<T>> clazz) { + throw new RuntimeException("Not Implemented"); + } + + @Override + public void close() throws IOException { + throw new RuntimeException("Not Implemented"); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/da069f03/blur-core/src/main/java/org/apache/blur/manager/command/ControllerCommandManager.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/command/ControllerCommandManager.java b/blur-core/src/main/java/org/apache/blur/manager/command/ControllerCommandManager.java index 318a851..8f93445 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/command/ControllerCommandManager.java +++ b/blur-core/src/main/java/org/apache/blur/manager/command/ControllerCommandManager.java @@ -1,8 +1,10 @@ package org.apache.blur.manager.command; import java.io.IOException; +import java.util.Map; import org.apache.blur.manager.command.primitive.BaseCommand; +import org.apache.blur.server.TableContext; /** * Licensed to the Apache Software Foundation (ASF) under one or more @@ -20,28 +22,62 @@ import org.apache.blur.manager.command.primitive.BaseCommand; * License for the specific language governing permissions and limitations under * the License. */ - +@SuppressWarnings("unchecked") public class ControllerCommandManager extends BaseCommandManager { public ControllerCommandManager(int threadCount) throws IOException { super(threadCount); } - public Response execute(String table, String commandName, Args args) { - ClusterContext context = createCommandContext(table); - BaseCommand command = getCommand(commandName); - - // For those commands that do not implement cluster command, run them in a base impl. - - throw new RuntimeException("Not Implemented"); + public Response execute(TableContext tableContext, String commandName, Args args) throws IOException { + ClusterContext context = createCommandContext(tableContext, args); + BaseCommand command = getCommandObject(commandName); + if (command == null) { + throw new IOException("Command with name [" + commandName + "] not found."); + } + // For those commands that do not implement cluster command, run them in a + // base impl. + + if (command instanceof ClusterCommand) { + return executeClusterCommand(context, command); + } else if (command instanceof IndexReadCombiningCommand) { + return executeIndexReadCombiningCommand(args, context, command); + } else if (command instanceof IndexReadCommand) { + return executeIndexReadCommand(args, context, command); + } else if (command instanceof IndexWriteCommand) { + return executeIndexWriteCommand(args, context, command); + } else { + throw new IOException("Command type of [" + command.getClass() + "] not supported."); + } + } + + private Response executeClusterCommand(ClusterContext context, BaseCommand command) { + ClusterCommand<Object> clusterCommand = (ClusterCommand<Object>) command; + Object object = clusterCommand.clusterExecute(context); + return Response.createNewAggregateResponse(object); + } + + private Response executeIndexWriteCommand(Args args, ClusterContext context, BaseCommand command) { + Class<? extends IndexWriteCommand<Object>> clazz = (Class<? extends IndexWriteCommand<Object>>) command.getClass(); + Object object = context.writeIndex(args, clazz); + return Response.createNewAggregateResponse(object); + } + + private Response executeIndexReadCommand(Args args, ClusterContext context, BaseCommand command) { + Class<? extends IndexReadCommand<Object>> clazz = (Class<? extends IndexReadCommand<Object>>) command.getClass(); + Map<Shard, Object> result = context.readIndexes(args, clazz); + return Response.createNewShardResponse(result); } - private BaseCommand getCommand(String commandName) { - throw new RuntimeException("Not Implemented"); + private Response executeIndexReadCombiningCommand(Args args, ClusterContext context, BaseCommand command) { + Class<? extends IndexReadCombiningCommand<Object, Object>> clazz = (Class<? extends IndexReadCombiningCommand<Object, Object>>) command + .getClass(); + Map<Server, Object> result = context.readServers(args, clazz); + return Response.createNewServerResponse(result); } - private ClusterContext createCommandContext(String table) { - throw new RuntimeException("Not Implemented"); + private ClusterContext createCommandContext(TableContext tableContext, Args args) { + return new ControllerClusterContext(tableContext, args); } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/da069f03/blur-core/src/main/java/org/apache/blur/manager/command/IndexWriteCommand.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/command/IndexWriteCommand.java b/blur-core/src/main/java/org/apache/blur/manager/command/IndexWriteCommand.java index 893638b..6ebc440 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/command/IndexWriteCommand.java +++ b/blur-core/src/main/java/org/apache/blur/manager/command/IndexWriteCommand.java @@ -18,9 +18,12 @@ package org.apache.blur.manager.command; import java.io.IOException; +import org.apache.blur.server.TableContext; import org.apache.lucene.index.IndexWriter; public interface IndexWriteCommand<T> { + + public abstract Shard route(Args args, TableContext context) throws IOException; public abstract T execute(IndexContext context, IndexWriter writer) throws IOException; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/da069f03/blur-core/src/main/java/org/apache/blur/manager/command/Response.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/command/Response.java b/blur-core/src/main/java/org/apache/blur/manager/command/Response.java index 1cf5b72..e26eeec 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/command/Response.java +++ b/blur-core/src/main/java/org/apache/blur/manager/command/Response.java @@ -21,13 +21,16 @@ import java.util.Map; public class Response { private final Map<Shard, Object> _shardResults; + private final Map<Server, Object> _serverResults; private final Object _serverResult; private final boolean _aggregatedResults; - private Response(Map<Shard, Object> shardResults, Object serverResult, boolean aggregatedResults) { + private Response(Map<Shard, Object> shardResults, Object serverResult, Map<Server, Object> serverResults, + boolean aggregatedResults) { _shardResults = shardResults; _serverResult = serverResult; _aggregatedResults = aggregatedResults; + _serverResults = serverResults; } public boolean isAggregatedResults() { @@ -42,11 +45,19 @@ public class Response { return _serverResult; } + public Map<Server, Object> getServerResults() { + return _serverResults; + } + public static Response createNewAggregateResponse(Object object) { - return new Response(null, object, true); + return new Response(null, object, null, true); + } + + public static Response createNewShardResponse(Map<Shard, Object> map) { + return new Response(map, null, null, false); } - public static Response createNewResponse(Map<Shard, Object> map) { - return new Response(map, null, false); + public static Response createNewServerResponse(Map<Server, Object> result) { + return new Response(null, null, result, true); } } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/da069f03/blur-core/src/main/java/org/apache/blur/manager/command/ShardCommandManager.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/command/ShardCommandManager.java b/blur-core/src/main/java/org/apache/blur/manager/command/ShardCommandManager.java index d48b6b6..5586a44 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/command/ShardCommandManager.java +++ b/blur-core/src/main/java/org/apache/blur/manager/command/ShardCommandManager.java @@ -61,7 +61,7 @@ public class ShardCommandManager extends BaseCommandManager { Object object = primitiveCommandAggregator.combine(results); return Response.createNewAggregateResponse(object); } - return Response.createNewResponse(results); + return Response.createNewShardResponse(results); } private Map<Shard, Object> executeReadWriteCommand(IndexWriteCommand<?> command, TableContext tableContext, Args args) { @@ -106,10 +106,6 @@ public class ShardCommandManager extends BaseCommandManager { return resultMap; } - private BaseCommand getCommandObject(String commandName) { - return _command.get(commandName); - } - static class ShardIndexContext extends IndexContext { private final TableContext _tableContext; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/da069f03/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 9c6120b..efb2f20 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 @@ -229,11 +229,15 @@ public class BlurControllerServer extends TableAdmin implements Iface { protected void tableContextWarmup() throws BlurException, TException { for (String table : tableList()) { LOG.debug("Warming the tablecontext for table [{0}]", table); - TableDescriptor describe = describe(table); - TableContext.create(describe); + getTableContext(table); } } + private TableContext getTableContext(String table) throws BlurException, TException { + TableDescriptor describe = describe(table); + return TableContext.create(describe); + } + private long getRandomDelay(long min, long max) { Random random = new Random(); return random.nextInt((int) (max - min)) + min; @@ -1500,7 +1504,8 @@ public class BlurControllerServer extends TableAdmin implements Iface { @Override public org.apache.blur.thrift.generated.Response execute(String table, String commandName, Arguments arguments) throws BlurException, TException { try { - Response response = _commandManager.execute(table, commandName, CommandUtil.convert(arguments)); + TableContext tableContext = getTableContext(table); + Response response = _commandManager.execute(tableContext, commandName, CommandUtil.convert(arguments)); return CommandUtil.convert(response); } catch (Exception e) { LOG.error("Unknown error while trying to execute command [{0}] for table [{1}]", e, commandName, table);
