-
Notifications
You must be signed in to change notification settings - Fork 24.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
The unfollow API changes a follower index into a regular index, so that it will accept write requests from clients. For the unfollow api to work the index follow needs to be stopped and the index needs to be closed. Closes #33931
- Loading branch information
Showing
11 changed files
with
408 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
117 changes: 117 additions & 0 deletions
117
.../plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/TransportUnfollowAction.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License; | ||
| * you may not use this file except in compliance with the Elastic License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.xpack.ccr.action; | ||
|
|
||
| import org.elasticsearch.action.ActionListener; | ||
| import org.elasticsearch.action.support.ActionFilters; | ||
| import org.elasticsearch.action.support.master.AcknowledgedResponse; | ||
| import org.elasticsearch.action.support.master.TransportMasterNodeAction; | ||
| import org.elasticsearch.cluster.ClusterState; | ||
| import org.elasticsearch.cluster.ClusterStateUpdateTask; | ||
| import org.elasticsearch.cluster.block.ClusterBlockException; | ||
| import org.elasticsearch.cluster.block.ClusterBlockLevel; | ||
| import org.elasticsearch.cluster.metadata.IndexMetaData; | ||
| import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver; | ||
| import org.elasticsearch.cluster.metadata.MetaData; | ||
| import org.elasticsearch.cluster.service.ClusterService; | ||
| import org.elasticsearch.common.inject.Inject; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.persistent.PersistentTasksCustomMetaData; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
| import org.elasticsearch.transport.TransportService; | ||
| import org.elasticsearch.xpack.ccr.Ccr; | ||
| import org.elasticsearch.xpack.ccr.CcrSettings; | ||
| import org.elasticsearch.xpack.core.ccr.action.UnfollowAction; | ||
|
|
||
| public class TransportUnfollowAction extends TransportMasterNodeAction<UnfollowAction.Request, AcknowledgedResponse> { | ||
|
|
||
| @Inject | ||
| public TransportUnfollowAction(Settings settings, TransportService transportService, ClusterService clusterService, | ||
| ThreadPool threadPool, ActionFilters actionFilters, | ||
| IndexNameExpressionResolver indexNameExpressionResolver) { | ||
| super(settings, UnfollowAction.NAME, transportService, clusterService, threadPool, actionFilters, | ||
| UnfollowAction.Request::new, indexNameExpressionResolver); | ||
| } | ||
|
|
||
| @Override | ||
| protected String executor() { | ||
| return ThreadPool.Names.SAME; | ||
| } | ||
|
|
||
| @Override | ||
| protected AcknowledgedResponse newResponse() { | ||
| return new AcknowledgedResponse(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void masterOperation(UnfollowAction.Request request, | ||
| ClusterState state, | ||
| ActionListener<AcknowledgedResponse> listener) throws Exception { | ||
| clusterService.submitStateUpdateTask("unfollow_action", new ClusterStateUpdateTask() { | ||
|
|
||
| @Override | ||
| public ClusterState execute(ClusterState current) throws Exception { | ||
| String followerIndex = request.getFollowerIndex(); | ||
| return unfollow(followerIndex, current); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(String source, Exception e) { | ||
| listener.onFailure(e); | ||
| } | ||
|
|
||
| @Override | ||
| public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) { | ||
| listener.onResponse(new AcknowledgedResponse(true)); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| @Override | ||
| protected ClusterBlockException checkBlock(UnfollowAction.Request request, ClusterState state) { | ||
| return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE); | ||
| } | ||
|
|
||
| static ClusterState unfollow(String followerIndex, ClusterState current) { | ||
| IndexMetaData followerIMD = current.metaData().index(followerIndex); | ||
|
|
||
| PersistentTasksCustomMetaData persistentTasks = current.metaData().custom(PersistentTasksCustomMetaData.TYPE); | ||
| if (persistentTasks != null) { | ||
| for (PersistentTasksCustomMetaData.PersistentTask<?> persistentTask : persistentTasks.tasks()) { | ||
| if (persistentTask.getTaskName().equals(ShardFollowTask.NAME)) { | ||
| ShardFollowTask shardFollowTask = (ShardFollowTask) persistentTask.getParams(); | ||
| if (shardFollowTask.getFollowShardId().getIndexName().equals(followerIndex)) { | ||
| throw new IllegalArgumentException("cannot convert the follower index [" + followerIndex + | ||
| "] to a non-follower, because it has not been paused"); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if (followerIMD.getState() != IndexMetaData.State.CLOSE) { | ||
| throw new IllegalArgumentException("cannot convert the follower index [" + followerIndex + | ||
| "] to a non-follower, because it has not been closed"); | ||
| } | ||
|
|
||
| IndexMetaData.Builder newIMD = IndexMetaData.builder(followerIMD); | ||
| // Remove index.xpack.ccr.following_index setting | ||
| Settings.Builder builder = Settings.builder(); | ||
| builder.put(followerIMD.getSettings()); | ||
| builder.remove(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey()); | ||
|
|
||
| newIMD.settings(builder); | ||
| // Remove ccr custom metadata | ||
| newIMD.removeCustom(Ccr.CCR_CUSTOM_METADATA_KEY); | ||
|
|
||
| MetaData newMetaData = MetaData.builder(current.metaData()) | ||
| .put(newIMD) | ||
| .build(); | ||
| return ClusterState.builder(current) | ||
| .metaData(newMetaData) | ||
| .build(); | ||
| } | ||
| } |
39 changes: 39 additions & 0 deletions
39
x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/rest/RestUnfollowAction.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License; | ||
| * you may not use this file except in compliance with the Elastic License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.xpack.ccr.rest; | ||
|
|
||
| import org.elasticsearch.client.node.NodeClient; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.rest.BaseRestHandler; | ||
| import org.elasticsearch.rest.RestController; | ||
| import org.elasticsearch.rest.RestRequest; | ||
| import org.elasticsearch.rest.action.RestToXContentListener; | ||
| import org.elasticsearch.xpack.core.ccr.action.UnfollowAction; | ||
|
|
||
| import java.io.IOException; | ||
|
|
||
| import static org.elasticsearch.xpack.core.ccr.action.UnfollowAction.INSTANCE; | ||
|
|
||
| public class RestUnfollowAction extends BaseRestHandler { | ||
|
|
||
| public RestUnfollowAction(Settings settings, RestController controller) { | ||
| super(settings); | ||
| controller.registerHandler(RestRequest.Method.POST, "/{index}/_ccr/unfollow", this); | ||
| } | ||
|
|
||
| @Override | ||
| public String getName() { | ||
| return "ccr_unfollow_action"; | ||
| } | ||
|
|
||
| @Override | ||
| protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException { | ||
| UnfollowAction.Request request = new UnfollowAction.Request(restRequest.param("index")); | ||
| return channel -> client.execute(INSTANCE, request, new RestToXContentListener<>(channel)); | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
103 changes: 103 additions & 0 deletions
103
...in/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/TransportUnfollowActionTests.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,103 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License; | ||
| * you may not use this file except in compliance with the Elastic License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.xpack.ccr.action; | ||
|
|
||
| import org.elasticsearch.Version; | ||
| import org.elasticsearch.cluster.ClusterName; | ||
| import org.elasticsearch.cluster.ClusterState; | ||
| import org.elasticsearch.cluster.metadata.IndexMetaData; | ||
| import org.elasticsearch.cluster.metadata.MetaData; | ||
| import org.elasticsearch.common.unit.TimeValue; | ||
| import org.elasticsearch.index.shard.ShardId; | ||
| import org.elasticsearch.persistent.PersistentTasksCustomMetaData; | ||
| import org.elasticsearch.test.ESTestCase; | ||
| import org.elasticsearch.xpack.ccr.Ccr; | ||
| import org.elasticsearch.xpack.ccr.CcrSettings; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.HashMap; | ||
|
|
||
| import static org.hamcrest.Matchers.equalTo; | ||
| import static org.hamcrest.Matchers.nullValue; | ||
|
|
||
| public class TransportUnfollowActionTests extends ESTestCase { | ||
|
|
||
| public void testUnfollow() { | ||
| IndexMetaData.Builder followerIndex = IndexMetaData.builder("follow_index") | ||
| .settings(settings(Version.CURRENT).put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true)) | ||
| .numberOfShards(1) | ||
| .numberOfReplicas(0) | ||
| .state(IndexMetaData.State.CLOSE) | ||
| .putCustom(Ccr.CCR_CUSTOM_METADATA_KEY, new HashMap<>()); | ||
|
|
||
| ClusterState current = ClusterState.builder(new ClusterName("cluster_name")) | ||
| .metaData(MetaData.builder() | ||
| .put(followerIndex) | ||
| .build()) | ||
| .build(); | ||
| ClusterState result = TransportUnfollowAction.unfollow("follow_index", current); | ||
|
|
||
| IndexMetaData resultIMD = result.metaData().index("follow_index"); | ||
| assertThat(resultIMD.getSettings().get(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey()), nullValue()); | ||
| assertThat(resultIMD.getCustomData(Ccr.CCR_CUSTOM_METADATA_KEY), nullValue()); | ||
| } | ||
|
|
||
| public void testUnfollowIndexOpen() { | ||
| IndexMetaData.Builder followerIndex = IndexMetaData.builder("follow_index") | ||
| .settings(settings(Version.CURRENT).put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true)) | ||
| .numberOfShards(1) | ||
| .numberOfReplicas(0) | ||
| .putCustom(Ccr.CCR_CUSTOM_METADATA_KEY, new HashMap<>()); | ||
|
|
||
| ClusterState current = ClusterState.builder(new ClusterName("cluster_name")) | ||
| .metaData(MetaData.builder() | ||
| .put(followerIndex) | ||
| .build()) | ||
| .build(); | ||
| Exception e = expectThrows(IllegalArgumentException.class, () -> TransportUnfollowAction.unfollow("follow_index", current)); | ||
| assertThat(e.getMessage(), | ||
| equalTo("cannot convert the follower index [follow_index] to a non-follower, because it has not been closed")); | ||
| } | ||
|
|
||
| public void testUnfollowRunningShardFollowTasks() { | ||
| IndexMetaData.Builder followerIndex = IndexMetaData.builder("follow_index") | ||
| .settings(settings(Version.CURRENT).put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), true)) | ||
| .numberOfShards(1) | ||
| .numberOfReplicas(0) | ||
| .state(IndexMetaData.State.CLOSE) | ||
| .putCustom(Ccr.CCR_CUSTOM_METADATA_KEY, new HashMap<>()); | ||
|
|
||
|
|
||
| ShardFollowTask params = new ShardFollowTask( | ||
| null, | ||
| new ShardId("follow_index", "", 0), | ||
| new ShardId("leader_index", "", 0), | ||
| 1024, | ||
| 1, | ||
| TransportResumeFollowAction.DEFAULT_MAX_BATCH_SIZE_IN_BYTES, | ||
| 1, | ||
| 10240, | ||
| TimeValue.timeValueMillis(10), | ||
| TimeValue.timeValueMillis(10), | ||
| "uuid", | ||
| Collections.emptyMap() | ||
| ); | ||
| PersistentTasksCustomMetaData.PersistentTask<?> task = | ||
| new PersistentTasksCustomMetaData.PersistentTask<>("id", ShardFollowTask.NAME, params, 0, null); | ||
|
|
||
| ClusterState current = ClusterState.builder(new ClusterName("cluster_name")) | ||
| .metaData(MetaData.builder() | ||
| .put(followerIndex) | ||
| .putCustom(PersistentTasksCustomMetaData.TYPE, new PersistentTasksCustomMetaData(0, Collections.singletonMap("id", task))) | ||
| .build()) | ||
| .build(); | ||
| Exception e = expectThrows(IllegalArgumentException.class, () -> TransportUnfollowAction.unfollow("follow_index", current)); | ||
| assertThat(e.getMessage(), | ||
| equalTo("cannot convert the follower index [follow_index] to a non-follower, because it has not been paused")); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.