RemoveObsoleteShardsOperation implements MasterOperation { @Override public List<OperationId> execute(MasterContext context, List<MasterOperation> runningOperations) throws Exception { InteractionProtocol protocol = context.getProtocol(); Collection<String> nodeShards = protocol.getNodeShards(_nodeName); List<String> obsoletShards = collectObsoleteShards(protocol, nodeShards, runningOperations); if (!obsoletShards.isEmpty()) { Log.info("found following shards obsolete on node " + _nodeName + ": " + obsoletShards); protocol.addNodeOperation(_nodeName, new ShardUndeployOperation(obsoletShards)); } return null; } RemoveObsoleteShardsOperation(String nodeName); String getNodeName(); @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 testMockRemove() throws Exception { String nodeName = "nodeA"; String someOldShard = AbstractIndexOperation.createShardName("someOldIndex", "someOldShard"); InteractionProtocol protocol = mock(InteractionProtocol.class); MasterQueue queue = mock(MasterQueue.class); MasterContext context = new MasterContext(protocol, Mocks.mockMaster(), new DefaultDistributionPolicy(), queue); when(protocol.getNodeShards(nodeName)).thenReturn(Arrays.asList(someOldShard)); RemoveObsoleteShardsOperation operation = new RemoveObsoleteShardsOperation(nodeName); operation.execute(context, EMPTY_LIST); ArgumentCaptor<NodeOperation> captor = ArgumentCaptor.forClass(NodeOperation.class); verify(protocol).addNodeOperation(eq(nodeName), captor.capture()); assertThat(captor.getValue(), instanceOf(ShardUndeployOperation.class)); ShardUndeployOperation undeployOperation = (ShardUndeployOperation) captor.getValue(); assertEquals(1, undeployOperation.getShardNames().size()); assertEquals(someOldShard, undeployOperation.getShardNames().iterator().next()); }
@Test public void testNotRemoveDeployingIndex() throws Exception { String nodeName = "nodeA"; String indexName = "someOldIndex"; String someOldShard = AbstractIndexOperation.createShardName(indexName, "someOldShard"); InteractionProtocol protocol = mock(InteractionProtocol.class); MasterQueue queue = mock(MasterQueue.class); MasterContext context = new MasterContext(protocol, Mocks.mockMaster(), new DefaultDistributionPolicy(), queue); when(protocol.getNodeShards(nodeName)).thenReturn(Arrays.asList(someOldShard)); RemoveObsoleteShardsOperation operation = new RemoveObsoleteShardsOperation(nodeName); operation.execute(context, new ArrayList<MasterOperation>(Arrays.asList(new IndexDeployOperation(indexName, "path", 1)))); verify(protocol, times(0)).addNodeOperation(eq(nodeName), (NodeOperation) notNull()); }