Skip to content

Commit c6c11f9

Browse files
tlrxtlrx
committed
Clean up ShardFollowTasks for deleted indices (elastic#44702)
Deleting a follower index does not delete its ShardFollowTasks, potentially leaving many persistent tasks in the cluster that cannot be allocated on nodes and unnecessary fill the logs. This commit adds a cluster state listener (ShardFollowTaskCleaner) that completes (with a failure) any persistent task that refers to a non existent follower index. I think that this bug has been introduced by elastic#34404: before this change the task would have been completed as failed and removed from the cluster state.
1 parent 26da573 commit c6c11f9

File tree

3 files changed

+125
-6
lines changed

3 files changed

+125
-6
lines changed

‎x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/Ccr.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@
5353
import org.elasticsearch.xpack.ccr.action.CcrRequests;
5454
import org.elasticsearch.xpack.ccr.action.ShardChangesAction;
5555
import org.elasticsearch.xpack.ccr.action.ShardFollowTask;
56+
import org.elasticsearch.xpack.ccr.action.ShardFollowTaskCleaner;
5657
import org.elasticsearch.xpack.ccr.action.ShardFollowTasksExecutor;
5758
import org.elasticsearch.xpack.ccr.action.TransportCcrStatsAction;
5859
import org.elasticsearch.xpack.ccr.action.TransportDeleteAutoFollowPatternAction;
@@ -184,6 +185,7 @@ public Collection<Object> createComponents(
184185
ccrLicenseChecker,
185186
restoreSourceService,
186187
new CcrRepositoryManager(settings, clusterService, client),
188+
new ShardFollowTaskCleaner(clusterService, threadPool, client),
187189
new AutoFollowCoordinator(
188190
settings,
189191
client,

‎x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTaskCleaner.java‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the Elastic License;
4+
* you may not use this file except in compliance with the Elastic License.
5+
*/
6+
7+
package org.elasticsearch.xpack.ccr.action;
8+
9+
import org.apache.logging.log4j.LogManager;
10+
import org.apache.logging.log4j.Logger;
11+
import org.elasticsearch.action.ActionListener;
12+
import org.elasticsearch.client.Client;
13+
import org.elasticsearch.cluster.ClusterChangedEvent;
14+
import org.elasticsearch.cluster.ClusterStateListener;
15+
import org.elasticsearch.cluster.metadata.MetaData;
16+
import org.elasticsearch.cluster.service.ClusterService;
17+
import org.elasticsearch.gateway.GatewayService;
18+
import org.elasticsearch.index.Index;
19+
import org.elasticsearch.index.IndexNotFoundException;
20+
import org.elasticsearch.persistent.CompletionPersistentTaskAction;
21+
import org.elasticsearch.persistent.PersistentTaskResponse;
22+
import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
23+
import org.elasticsearch.threadpool.ThreadPool;
24+
25+
/**
26+
* A {@link ClusterStateListener} that completes any {@link ShardFollowTask} which concerns a deleted index.
27+
*/
28+
public class ShardFollowTaskCleaner implements ClusterStateListener {
29+
30+
private static final Logger logger = LogManager.getLogger(ShardFollowTaskCleaner.class);
31+
32+
private final ThreadPool threadPool;
33+
private final Client client;
34+
35+
public ShardFollowTaskCleaner(final ClusterService clusterService, final ThreadPool threadPool, final Client client) {
36+
this.threadPool = threadPool;
37+
this.client = client;
38+
clusterService.addListener(this);
39+
}
40+
41+
@Override
42+
public void clusterChanged(final ClusterChangedEvent event) {
43+
if (event.state().blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK)) {
44+
return;
45+
}
46+
if (event.localNodeMaster() == false) {
47+
return;
48+
}
49+
50+
MetaData metaData = event.state().metaData();
51+
PersistentTasksCustomMetaData persistentTasksMetaData = metaData.custom(PersistentTasksCustomMetaData.TYPE);
52+
if (persistentTasksMetaData == null) {
53+
return;
54+
}
55+
for (PersistentTasksCustomMetaData.PersistentTask<?> persistentTask : persistentTasksMetaData.tasks()) {
56+
if (ShardFollowTask.NAME.equals(persistentTask.getTaskName()) == false) {
57+
// this task is not a shard follow task
58+
continue;
59+
}
60+
ShardFollowTask shardFollowTask = (ShardFollowTask) persistentTask.getParams();
61+
Index followerIndex = shardFollowTask.getFollowShardId().getIndex();
62+
if (metaData.index(followerIndex) != null) {
63+
// the index exists, do not clean this persistent task
64+
continue;
65+
}
66+
IndexNotFoundException e = new IndexNotFoundException(followerIndex);
67+
CompletionPersistentTaskAction.Request request =
68+
new CompletionPersistentTaskAction.Request(persistentTask.getId(), persistentTask.getAllocationId(), e);
69+
threadPool.generic().submit(() -> {
70+
client.execute(CompletionPersistentTaskAction.INSTANCE, request, new ActionListener<PersistentTaskResponse>() {
71+
72+
@Override
73+
public void onResponse(PersistentTaskResponse persistentTaskResponse) {
74+
logger.debug("task [{}] cleaned up", persistentTask.getId());
75+
}
76+
77+
@Override
78+
public void onFailure(Exception e) {
79+
logger.warn("failed to clean up task [{}]", persistentTask.getId());
80+
}
81+
});
82+
});
83+
}
84+
}
85+
}

‎x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/IndexFollowingIT.java‎

Lines changed: 38 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -762,13 +762,14 @@ public void testDeleteFollowerIndex() throws Exception {
762762
StatsResponses response = followerClient().execute(FollowStatsAction.INSTANCE, new StatsRequest()).actionGet();
763763
assertThat(response.getNodeFailures(), empty());
764764
assertThat(response.getTaskFailures(), empty());
765-
assertThat(response.getStatsResponses(), hasSize(1));
766-
assertThat(response.getStatsResponses().get(0).status().failedWriteRequests(), greaterThanOrEqualTo(1L));
767-
ElasticsearchException fatalException = response.getStatsResponses().get(0).status().getFatalException();
768-
assertThat(fatalException, notNullValue());
769-
assertThat(fatalException.getMessage(), equalTo("no such index [index2]"));
765+
if (response.getStatsResponses().isEmpty() == false) {
766+
assertThat(response.getStatsResponses(), hasSize(1));
767+
assertThat(response.getStatsResponses().get(0).status().failedWriteRequests(), greaterThanOrEqualTo(1L));
768+
ElasticsearchException fatalException = response.getStatsResponses().get(0).status().getFatalException();
769+
assertThat(fatalException, notNullValue());
770+
assertThat(fatalException.getMessage(), equalTo("no such index [index2]"));
771+
}
770772
});
771-
pauseFollow("index2");
772773
ensureNoCcrTasks();
773774
}
774775

@@ -1321,6 +1322,37 @@ public void testUpdateRemoteConfigsDuringFollowing() throws Exception {
13211322
}
13221323
}
13231324

1325+
public void testCleanUpShardFollowTasksForDeletedIndices() throws Exception {
1326+
final int numberOfShards = randomIntBetween(1, 10);
1327+
assertAcked(leaderClient().admin().indices().prepareCreate("index1")
1328+
.setSettings(Settings.builder()
1329+
.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
1330+
.put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, numberOfShards)
1331+
.put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, randomIntBetween(0, 1))
1332+
.build()));
1333+
1334+
final PutFollowAction.Request followRequest = putFollow("index1", "index2");
1335+
followerClient().execute(PutFollowAction.INSTANCE, followRequest).get();
1336+
1337+
leaderClient().prepareIndex("index1", "doc", "1").setSource("{}", XContentType.JSON).get();
1338+
assertBusy(() -> assertThat(followerClient().prepareSearch("index2").get().getHits().getTotalHits().value, equalTo(1L)));
1339+
1340+
assertBusy(() -> {
1341+
String action = ShardFollowTask.NAME + "[c]";
1342+
ListTasksResponse listTasksResponse = followerClient().admin().cluster().prepareListTasks().setActions(action).get();
1343+
assertThat(listTasksResponse.getTasks(), hasSize(numberOfShards));
1344+
});
1345+
1346+
assertAcked(followerClient().admin().indices().prepareDelete("index2"));
1347+
1348+
assertBusy(() -> {
1349+
String action = ShardFollowTask.NAME + "[c]";
1350+
ListTasksResponse listTasksResponse = followerClient().admin().cluster().prepareListTasks().setActions(action).get();
1351+
assertThat(listTasksResponse.getTasks(), hasSize(0));
1352+
});
1353+
ensureNoCcrTasks();
1354+
}
1355+
13241356
private long getFollowTaskSettingsVersion(String followerIndex) {
13251357
long settingsVersion = -1L;
13261358
for (ShardFollowNodeTaskStatus status : getFollowTaskStatuses(followerIndex)) {

0 commit comments

Comments
 (0)