BalanceIndexOperation extends AbstractIndexOperation { @Override public List<OperationId> execute(MasterContext context, List<MasterOperation> runningOperations) throws Exception { InteractionProtocol protocol = context.getProtocol(); IndexMetaData indexMD = protocol.getIndexMD(_indexName); if (indexMD == null) { LOG.info("skip balancing for index '" + _indexName + "' cause it is already undeployed"); return null; } if (!canAndShouldRegulateReplication(protocol, indexMD)) { LOG.info("skip balancing for index '" + _indexName + "' cause there is no possible optimization"); return null; } try { FileSystem fileSystem = context.getFileSystem(indexMD); Path path = new Path(indexMD.getPath()); if (!fileSystem.exists(path)) { LOG.warn("skip balancing for index '" + _indexName + "' cause source '" + path + "' does not exists anymore"); return null; } } catch (Exception e) { LOG.error("skip balancing of index '" + _indexName + "' cause failed to access source '" + indexMD.getPath() + "'", e); return null; } LOG.info("balancing shards for index '" + _indexName + "'"); try { List<OperationId> operationIds = distributeIndexShards(context, indexMD, protocol.getLiveNodes(), runningOperations); return operationIds; } catch (Exception e) { ExceptionUtil.rethrowInterruptedException(e); LOG.error("failed to deploy balance " + _indexName, e); handleMasterDeployException(protocol, indexMD, e); return null; } } BalanceIndexOperation(String indexName); @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 testBalanceUnderreplicatedIndex() throws Exception { List<Node> nodes = Mocks.mockNodes(2); List<NodeQueue> nodeQueues = Mocks.publisNodes(_protocol, nodes); deployIndexWithError(); for (NodeQueue nodeqQueue : nodeQueues) { assertEquals(1, nodeqQueue.size()); } publisShards(nodes, nodeQueues); BalanceIndexOperation balanceOperation = new BalanceIndexOperation(_indexName); balanceOperation.execute(_context, EMPTY_LIST); for (NodeQueue nodeqQueue : nodeQueues) { assertEquals(0, nodeqQueue.size()); } Node node3 = Mocks.mockNode(); NodeQueue nodeQueue3 = Mocks.publisNode(_protocol, node3); assertEquals(0, nodeQueue3.size()); balanceOperation.execute(_context, EMPTY_LIST); for (NodeQueue nodeqQueue : nodeQueues) { assertEquals(0, nodeqQueue.size()); } assertEquals(1, nodeQueue3.size()); }
@Test public void testBalanceOverreplicatedIndex() throws Exception { List<Node> nodes = Mocks.mockNodes(3); List<NodeQueue> nodeQueues = Mocks.publisNodes(_protocol, nodes); deployIndexWithError(); for (NodeQueue nodeqQueue : nodeQueues) { assertEquals(1, nodeqQueue.size()); } publisShards(nodes, nodeQueues); BalanceIndexOperation balanceOperation = new BalanceIndexOperation(_indexName); balanceOperation.execute(_context, EMPTY_LIST); for (NodeQueue nodeqQueue : nodeQueues) { assertEquals(0, nodeqQueue.size()); } IndexMetaData indexMD = _protocol.getIndexMD(_indexName); indexMD.setReplicationLevel(2); _protocol.updateIndexMD(indexMD); balanceOperation.execute(_context, EMPTY_LIST); for (NodeQueue nodeqQueue : nodeQueues) { assertEquals(1, nodeqQueue.size()); } }
@Test public void testStopBalance_WhenSourceFileDoesNotExistAnymore() throws Exception { List<Node> nodes = Mocks.mockNodes(2); List<NodeQueue> nodeQueues = Mocks.publisNodes(_protocol, nodes); deployIndexWithError(); for (NodeQueue nodeqQueue : nodeQueues) { assertEquals(1, nodeqQueue.size()); } publisShards(nodes, nodeQueues); Node node3 = Mocks.mockNode(); NodeQueue nodeQueue3 = Mocks.publisNode(_protocol, node3); assertEquals(0, nodeQueue3.size()); BalanceIndexOperation balanceOperation = new BalanceIndexOperation(_indexName); FileSystem fileSystem = mock(FileSystem.class); when(fileSystem.exists(any(Path.class))).thenReturn(false); MasterContext spiedContext = spy(_context); doReturn(fileSystem).when(spiedContext).getFileSystem(any(IndexMetaData.class)); List<OperationId> nodeOperations = balanceOperation.execute(spiedContext, EMPTY_LIST); assertEquals(null, nodeOperations); }
@Test public void testStopBalance_CantAccessSourceFile() throws Exception { List<Node> nodes = Mocks.mockNodes(2); List<NodeQueue> nodeQueues = Mocks.publisNodes(_protocol, nodes); deployIndexWithError(); for (NodeQueue nodeqQueue : nodeQueues) { assertEquals(1, nodeqQueue.size()); } publisShards(nodes, nodeQueues); Node node3 = Mocks.mockNode(); NodeQueue nodeQueue3 = Mocks.publisNode(_protocol, node3); assertEquals(0, nodeQueue3.size()); BalanceIndexOperation balanceOperation = new BalanceIndexOperation(_indexName); MasterContext spiedContext = spy(_context); doThrow(new RuntimeException("test-exception")).when(spiedContext).getFileSystem(any(IndexMetaData.class)); List<OperationId> nodeOperations = balanceOperation.execute(spiedContext, EMPTY_LIST); assertEquals(null, nodeOperations); }