IndexDeployOperation extends AbstractIndexOperation { @Override public List<OperationId> execute(MasterContext context, List<MasterOperation> runningOperations) throws Exception, InterruptedException { InteractionProtocol protocol = context.getProtocol(); try { _indexMD.getShards().addAll(readShardsFromFs(_indexName, _indexPath)); LOG.info("Found shards '" + _indexMD.getShards() + "' for index '" + _indexName + "'"); List<OperationId> operationIds = distributeIndexShards(context, _indexMD, protocol.getLiveNodes(), runningOperations); return operationIds; } catch (Exception e) { ExceptionUtil.rethrowInterruptedException(e); LOG.error("failed to deploy index " + _indexName, e); protocol.publishIndex(_indexMD); handleMasterDeployException(protocol, _indexMD, e); return null; } } IndexDeployOperation(String indexName, String indexPath, int replicationLevel); String getIndexName(); String getIndexPath(); int getReplicationLevel(); @Override List<OperationId> execute(MasterContext context, List<MasterOperation> runningOperations); @Override void nodeOperationsComplete(MasterContext context, List<OperationResult> results); @Override ExecutionInstruction getExecutionInstruction(List<MasterOperation> runningOperations); @Override String toString();  }
@Test public void testDeployError_NoNodes() throws Exception { IndexDeployOperation deployCommand = new IndexDeployOperation(_indexName, _indexPath, 3); deployCommand.execute(_context, EMPTY_LIST); checkDeployError(ErrorType.NO_NODES_AVAILIBLE, _shardCount); }
@Test public void testDeployError_IndexNotAccessable() throws Exception { IndexDeployOperation deployCommand = new IndexDeployOperation(_indexName, "wrongIndexPath", 3); deployCommand.execute(_context, EMPTY_LIST); checkDeployError(ErrorType.INDEX_NOT_ACCESSIBLE, 0); }
@Test public void testDeployRespectsCurrentRunningDeployments() throws Exception { List<Node> nodes = Mocks.mockNodes(2 * _shardCount); List<NodeQueue> nodeQueues = Mocks.publisNodes(_protocol, nodes); int replicationLevel = 1; IndexDeployOperation operation1 = new IndexDeployOperation(_indexName, _indexPath, replicationLevel); operation1.execute(_context, EMPTY_LIST); List<MasterOperation> runningOps = new ArrayList<MasterOperation>(); runningOps.add(operation1); IndexDeployOperation operation2 = new IndexDeployOperation(_indexName + "2", _indexPath, replicationLevel); operation2.execute(_context, runningOps); List<Integer> nodeQueueSizes = new ArrayList<Integer>(); for (NodeQueue nodeQueue : nodeQueues) { nodeQueueSizes.add(nodeQueue.size()); } for (Integer integer : nodeQueueSizes) { assertEquals("unequal shard distribution: " + nodeQueueSizes, 1, integer.intValue()); } }