ElasticIndexCleaner implements Runnable { public void run() { try { NodeState root = nodeStore.getRoot(); GetIndexRequest getIndexRequest = new GetIndexRequest(elasticConnection.getIndexPrefix() + "*") .indicesOptions(IndicesOptions.lenientExpandOpen()); String[] remoteIndices = elasticConnection.getClient().indices() .get(getIndexRequest, RequestOptions.DEFAULT).getIndices(); if (remoteIndices == null || remoteIndices.length == 0) { LOG.debug("No remote index found with prefix {}", indexPrefix); return; } List<String> externallyDeletedIndices = danglingRemoteIndicesMap.keySet().stream(). filter(index -> Arrays.stream(remoteIndices).noneMatch(remoteIndex -> remoteIndex.equals(index))).collect(Collectors.toList()); externallyDeletedIndices.forEach(danglingRemoteIndicesMap::remove); Set<String> existingIndices = new HashSet<>(); root.getChildNode(INDEX_DEFINITIONS_NAME).getChildNodeEntries().forEach(childNodeEntry -> { PropertyState typeProperty = childNodeEntry.getNodeState().getProperty(IndexConstants.TYPE_PROPERTY_NAME); String typeValue = typeProperty != null ? typeProperty.getValue(Type.STRING) : ""; if (typeValue.equals(ElasticIndexDefinition.TYPE_ELASTICSEARCH) || typeValue.equals("disabled")) { String indexPath = "/" + INDEX_DEFINITIONS_NAME + "/" + childNodeEntry.getName(); String remoteIndexName = ElasticIndexNameHelper.getRemoteIndexName(indexPrefix, childNodeEntry.getNodeState(), indexPath); if (remoteIndexName != null) { existingIndices.add(remoteIndexName); } } }); List<String> indicesToDelete = new ArrayList<>(); for (String remoteIndexName : remoteIndices) { if (!existingIndices.contains(remoteIndexName)) { Long curTime = System.currentTimeMillis(); Long oldTime = danglingRemoteIndicesMap.putIfAbsent(remoteIndexName, curTime); if (threshold == 0 || (oldTime != null && curTime - oldTime >= TimeUnit.SECONDS.toMillis(threshold))) { indicesToDelete.add(remoteIndexName); danglingRemoteIndicesMap.remove(remoteIndexName); } } else { danglingRemoteIndicesMap.remove(remoteIndexName); } } DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(indicesToDelete.toArray(new String[]{})); if (deleteIndexRequest.indices() != null && deleteIndexRequest.indices().length > 0) { String indexString = Arrays.toString(deleteIndexRequest.indices()); AcknowledgedResponse acknowledgedResponse = elasticConnection.getClient().indices().delete(deleteIndexRequest, RequestOptions.DEFAULT); LOG.info("Deleting remote indices {}", indexString); if (!acknowledgedResponse.isAcknowledged()) { LOG.error("Could not delete remote indices " + indexString); } } } catch (IOException e) { LOG.error("Could not delete remote indices", e); } } ElasticIndexCleaner(ElasticConnection elasticConnection, NodeStore nodeStore, int thresholdInSeconds); void run();  }
@Test public void testIndexDeletion() throws Exception { String indexId1 = createIndexAndContentNode("propa", "test1"); String indexId2 = createIndexAndContentNode("propb", "test2"); String indexId3 = createIndexAndContentNode("propc", "test3"); String indexPath1 = "/" + INDEX_DEFINITIONS_NAME + "/" + indexId1; String indexPath2 = "/" + INDEX_DEFINITIONS_NAME + "/" + indexId2; String indexPath3 = "/" + INDEX_DEFINITIONS_NAME + "/" + indexId3; NodeState oakIndex = nodeStore.getRoot().getChildNode(INDEX_DEFINITIONS_NAME); NodeState indexState1 = oakIndex.getChildNode(indexId1); NodeState indexState2 = oakIndex.getChildNode(indexId2); NodeState indexState3 = oakIndex.getChildNode(indexId3); root.refresh(); root.getTree(indexPath1).remove(); root.getTree(indexPath2).remove(); root.commit(); oakIndex = nodeStore.getRoot().getChildNode(INDEX_DEFINITIONS_NAME); assertFalse(oakIndex.getChildNode(indexId1).exists()); assertFalse(oakIndex.getChildNode(indexId2).exists()); ElasticIndexCleaner cleaner = new ElasticIndexCleaner(esConnection, nodeStore, 5); cleaner.run(); String remoteIndexName1 = ElasticIndexNameHelper.getRemoteIndexName(esConnection.getIndexPrefix(), indexState1, indexPath1); String remoteIndexName2 = ElasticIndexNameHelper.getRemoteIndexName(esConnection.getIndexPrefix(), indexState2, indexPath2); String remoteIndexName3 = ElasticIndexNameHelper.getRemoteIndexName(esConnection.getIndexPrefix(), indexState3, indexPath3); assertTrue(esConnection.getClient().indices().exists(new GetIndexRequest(remoteIndexName1), RequestOptions.DEFAULT)); assertTrue(esConnection.getClient().indices().exists(new GetIndexRequest(remoteIndexName2), RequestOptions.DEFAULT)); assertTrue(esConnection.getClient().indices().exists(new GetIndexRequest(remoteIndexName3), RequestOptions.DEFAULT)); Thread.sleep(5000); cleaner.run(); assertFalse(esConnection.getClient().indices().exists(new GetIndexRequest(remoteIndexName1), RequestOptions.DEFAULT)); assertFalse(esConnection.getClient().indices().exists(new GetIndexRequest(remoteIndexName2), RequestOptions.DEFAULT)); assertTrue(esConnection.getClient().indices().exists(new GetIndexRequest(remoteIndexName3), RequestOptions.DEFAULT)); }
@Test public void preventDisabledIndexDeletion() throws Exception { int indexDeletionThresholdTime = 5; String indexId = createIndexAndContentNode("propa", "test1"); String indexPath = "/" + INDEX_DEFINITIONS_NAME + "/" + indexId; NodeState oakIndex = nodeStore.getRoot().getChildNode(INDEX_DEFINITIONS_NAME); NodeState indexState = oakIndex.getChildNode(indexId); indexState.builder().remove(); root.refresh(); root.getTree(indexPath).setProperty("type", "disabled"); root.commit(); ElasticIndexCleaner cleaner = new ElasticIndexCleaner(esConnection, nodeStore, indexDeletionThresholdTime); cleaner.run(); String remoteIndexName = ElasticIndexNameHelper.getRemoteIndexName(esConnection.getIndexPrefix(), indexState, indexPath); assertTrue(esConnection.getClient().indices().exists(new GetIndexRequest(remoteIndexName), RequestOptions.DEFAULT)); Thread.sleep(TimeUnit.SECONDS.toMillis(indexDeletionThresholdTime)); cleaner.run(); assertTrue(esConnection.getClient().indices().exists(new GetIndexRequest(remoteIndexName), RequestOptions.DEFAULT)); }