[CCR] Added HLRC support for pause follow API (#35216)
* Moved `AcknowledgedResponse` to core package * Made AcknowledgedResponse not abstract and provided a default parser, so that in cases when the field name is not overwritten then there is no need for a subclass. Relates to #33824
This commit is contained in:
Martijn van Groningen
GitHub
parent
e572a21c4b
commit
021f80517f
|
|
@ -0,0 +1,86 @@
|
|||
/*
|
||||
* Licensed to Elasticsearch under one or more contributor
|
||||
* license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright
|
||||
* ownership. Elasticsearch licenses this file to you under
|
||||
* the Apache License, Version 2.0 (the "License"); you may
|
||||
* not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.client;
|
||||
|
||||
import org.elasticsearch.action.ActionListener;
|
||||
import org.elasticsearch.client.ccr.PauseFollowRequest;
|
||||
import org.elasticsearch.client.core.AcknowledgedResponse;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Collections;
|
||||
|
||||
/**
|
||||
* A wrapper for the {@link RestHighLevelClient} that provides methods for
|
||||
* accessing the Elastic ccr related methods
|
||||
* <p>
|
||||
* See the <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-apis.html">
|
||||
* X-Pack Rollup APIs on elastic.co</a> for more information.
|
||||
*/
|
||||
public final class CcrClient {
|
||||
|
||||
private final RestHighLevelClient restHighLevelClient;
|
||||
|
||||
CcrClient(RestHighLevelClient restHighLevelClient) {
|
||||
this.restHighLevelClient = restHighLevelClient;
|
||||
}
|
||||
|
||||
/**
|
||||
* Instructs a follower index the pause the following of a leader index.
|
||||
*
|
||||
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-pause-follow.html">
|
||||
* the docs</a> for more.
|
||||
*
|
||||
* @param request the request
|
||||
* @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
|
||||
* @return the response
|
||||
* @throws IOException in case there is a problem sending the request or parsing back the response
|
||||
*/
|
||||
public AcknowledgedResponse pauseFollow(PauseFollowRequest request, RequestOptions options) throws IOException {
|
||||
return restHighLevelClient.performRequestAndParseEntity(
|
||||
request,
|
||||
CcrRequestConverters::pauseFollow,
|
||||
options,
|
||||
AcknowledgedResponse::fromXContent,
|
||||
Collections.emptySet()
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Asynchronously instruct a follower index the pause the following of a leader index.
|
||||
*
|
||||
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-pause-follow.html">
|
||||
* the docs</a> for more.
|
||||
*
|
||||
* @param request the request
|
||||
* @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
|
||||
*/
|
||||
public void pauseFollowAsync(PauseFollowRequest request,
|
||||
RequestOptions options,
|
||||
ActionListener<AcknowledgedResponse> listener) {
|
||||
restHighLevelClient.performRequestAsyncAndParseEntity(
|
||||
request,
|
||||
CcrRequestConverters::pauseFollow,
|
||||
options,
|
||||
AcknowledgedResponse::fromXContent,
|
||||
listener,
|
||||
Collections.emptySet());
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,35 @@
|
|||
/*
|
||||
* Licensed to Elasticsearch under one or more contributor
|
||||
* license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright
|
||||
* ownership. Elasticsearch licenses this file to you under
|
||||
* the Apache License, Version 2.0 (the "License"); you may
|
||||
* not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.client;
|
||||
|
||||
import org.apache.http.client.methods.HttpPost;
|
||||
import org.elasticsearch.client.ccr.PauseFollowRequest;
|
||||
|
||||
final class CcrRequestConverters {
|
||||
|
||||
static Request pauseFollow(PauseFollowRequest pauseFollowRequest) {
|
||||
String endpoint = new RequestConverters.EndpointBuilder()
|
||||
.addPathPart(pauseFollowRequest.getFollowerIndex())
|
||||
.addPathPartAsIs("_ccr", "pause_follow")
|
||||
.build();
|
||||
return new Request(HttpPost.METHOD_NAME, endpoint);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -228,6 +228,7 @@ public class RestHighLevelClient implements Closeable {
|
|||
private final SecurityClient securityClient = new SecurityClient(this);
|
||||
private final IndexLifecycleClient ilmClient = new IndexLifecycleClient(this);
|
||||
private final RollupClient rollupClient = new RollupClient(this);
|
||||
private final CcrClient ccrClient = new CcrClient(this);
|
||||
|
||||
/**
|
||||
* Creates a {@link RestHighLevelClient} given the low level {@link RestClientBuilder} that allows to build the
|
||||
|
|
@ -321,6 +322,20 @@ public class RestHighLevelClient implements Closeable {
|
|||
return rollupClient;
|
||||
}
|
||||
|
||||
/**
|
||||
* Provides methods for accessing the Elastic Licensed CCR APIs that
|
||||
* are shipped with the Elastic Stack distribution of Elasticsearch. All of
|
||||
* these APIs will 404 if run against the OSS distribution of Elasticsearch.
|
||||
* <p>
|
||||
* See the <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-api.html">
|
||||
* CCR APIs on elastic.co</a> for more information.
|
||||
*
|
||||
* @return the client wrapper for making CCR API calls
|
||||
*/
|
||||
public final CcrClient ccr() {
|
||||
return ccrClient;
|
||||
}
|
||||
|
||||
/**
|
||||
* Provides a {@link TasksClient} which can be used to access the Tasks API.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -0,0 +1,37 @@
|
|||
/*
|
||||
* Licensed to Elasticsearch under one or more contributor
|
||||
* license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright
|
||||
* ownership. Elasticsearch licenses this file to you under
|
||||
* the Apache License, Version 2.0 (the "License"); you may
|
||||
* not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.client.ccr;
|
||||
|
||||
import org.elasticsearch.client.Validatable;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
public final class PauseFollowRequest implements Validatable {
|
||||
|
||||
private final String followerIndex;
|
||||
|
||||
public PauseFollowRequest(String followerIndex) {
|
||||
this.followerIndex = Objects.requireNonNull(followerIndex);
|
||||
}
|
||||
|
||||
public String getFollowerIndex() {
|
||||
return followerIndex;
|
||||
}
|
||||
}
|
||||
|
|
@ -17,13 +17,14 @@
|
|||
* under the License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.client.rollup;
|
||||
package org.elasticsearch.client.core;
|
||||
|
||||
import org.elasticsearch.common.ParseField;
|
||||
import org.elasticsearch.common.xcontent.ConstructingObjectParser;
|
||||
import org.elasticsearch.common.xcontent.ToXContent;
|
||||
import org.elasticsearch.common.xcontent.ToXContentObject;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Objects;
|
||||
|
|
@ -31,9 +32,12 @@ import java.util.function.Function;
|
|||
|
||||
import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
|
||||
|
||||
public abstract class AcknowledgedResponse implements ToXContentObject {
|
||||
public class AcknowledgedResponse implements ToXContentObject {
|
||||
|
||||
protected static final String PARSE_FIELD_NAME = "acknowledged";
|
||||
private static final ConstructingObjectParser<AcknowledgedResponse, Void> PARSER = AcknowledgedResponse
|
||||
.generateParser("acknowledged_response", AcknowledgedResponse::new, AcknowledgedResponse.PARSE_FIELD_NAME);
|
||||
|
||||
private final boolean acknowledged;
|
||||
|
||||
public AcknowledgedResponse(final boolean acknowledged) {
|
||||
|
|
@ -50,6 +54,10 @@ public abstract class AcknowledgedResponse implements ToXContentObject {
|
|||
return p;
|
||||
}
|
||||
|
||||
public static AcknowledgedResponse fromXContent(final XContentParser parser) throws IOException {
|
||||
return PARSER.parse(parser, null);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o) {
|
||||
|
|
@ -19,6 +19,7 @@
|
|||
|
||||
package org.elasticsearch.client.rollup;
|
||||
|
||||
import org.elasticsearch.client.core.AcknowledgedResponse;
|
||||
import org.elasticsearch.common.xcontent.ConstructingObjectParser;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@
|
|||
*/
|
||||
package org.elasticsearch.client.rollup;
|
||||
|
||||
import org.elasticsearch.client.core.AcknowledgedResponse;
|
||||
import org.elasticsearch.common.xcontent.ConstructingObjectParser;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@
|
|||
|
||||
package org.elasticsearch.client.rollup;
|
||||
|
||||
import org.elasticsearch.client.core.AcknowledgedResponse;
|
||||
import org.elasticsearch.common.xcontent.ConstructingObjectParser;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
|
||||
|
|
|
|||
|
|
@ -785,7 +785,8 @@ public class RestHighLevelClientTests extends ESTestCase {
|
|||
apiName.startsWith("graph.") == false &&
|
||||
apiName.startsWith("migration.") == false &&
|
||||
apiName.startsWith("security.") == false &&
|
||||
apiName.startsWith("index_lifecycle.") == false) {
|
||||
apiName.startsWith("index_lifecycle.") == false &&
|
||||
apiName.startsWith("ccr.") == false) {
|
||||
apiNotFound.add(apiName);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,43 @@
|
|||
/*
|
||||
* Licensed to Elasticsearch under one or more contributor
|
||||
* license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright
|
||||
* ownership. Elasticsearch licenses this file to you under
|
||||
* the Apache License, Version 2.0 (the "License"); you may
|
||||
* not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
*/
|
||||
package org.elasticsearch.client.core;
|
||||
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
import org.elasticsearch.test.AbstractXContentTestCase;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
public class AcknowledgedResponseTests extends AbstractXContentTestCase<AcknowledgedResponse> {
|
||||
|
||||
@Override
|
||||
protected AcknowledgedResponse createTestInstance() {
|
||||
return new AcknowledgedResponse(randomBoolean());
|
||||
}
|
||||
|
||||
@Override
|
||||
protected AcknowledgedResponse doParseInstance(XContentParser parser) throws IOException {
|
||||
return AcknowledgedResponse.fromXContent(parser);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean supportsUnknownFields() {
|
||||
return false;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,139 @@
|
|||
/*
|
||||
* Licensed to Elasticsearch under one or more contributor
|
||||
* license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright
|
||||
* ownership. Elasticsearch licenses this file to you under
|
||||
* the Apache License, Version 2.0 (the "License"); you may
|
||||
* not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.client.documentation;
|
||||
|
||||
import org.apache.http.util.EntityUtils;
|
||||
import org.elasticsearch.action.ActionListener;
|
||||
import org.elasticsearch.action.LatchedActionListener;
|
||||
import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
|
||||
import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsResponse;
|
||||
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
|
||||
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
|
||||
import org.elasticsearch.client.ESRestHighLevelClientTestCase;
|
||||
import org.elasticsearch.client.Request;
|
||||
import org.elasticsearch.client.RequestOptions;
|
||||
import org.elasticsearch.client.Response;
|
||||
import org.elasticsearch.client.RestHighLevelClient;
|
||||
import org.elasticsearch.client.ccr.PauseFollowRequest;
|
||||
import org.elasticsearch.client.core.AcknowledgedResponse;
|
||||
import org.elasticsearch.common.xcontent.XContentHelper;
|
||||
import org.elasticsearch.common.xcontent.json.JsonXContent;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Collections;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import static org.hamcrest.Matchers.equalTo;
|
||||
import static org.hamcrest.Matchers.is;
|
||||
|
||||
public class CCRDocumentationIT extends ESRestHighLevelClientTestCase {
|
||||
|
||||
public void testPauseFollow() throws Exception {
|
||||
RestHighLevelClient client = highLevelClient();
|
||||
|
||||
{
|
||||
// Configure local cluster as remote cluster:
|
||||
|
||||
// TODO: replace with nodes info highlevel rest client code when it is available:
|
||||
final Request request = new Request("GET", "/_nodes");
|
||||
Map<?, ?> nodesResponse = (Map<?, ?>) toMap(client().performRequest(request)).get("nodes");
|
||||
// Select node info of first node (we don't know the node id):
|
||||
nodesResponse = (Map<?, ?>) nodesResponse.get(nodesResponse.keySet().iterator().next());
|
||||
String transportAddress = (String) nodesResponse.get("transport_address");
|
||||
|
||||
ClusterUpdateSettingsRequest updateSettingsRequest = new ClusterUpdateSettingsRequest();
|
||||
updateSettingsRequest.transientSettings(Collections.singletonMap("cluster.remote.local.seeds", transportAddress));
|
||||
ClusterUpdateSettingsResponse updateSettingsResponse =
|
||||
client.cluster().putSettings(updateSettingsRequest, RequestOptions.DEFAULT);
|
||||
assertThat(updateSettingsResponse.isAcknowledged(), is(true));
|
||||
}
|
||||
{
|
||||
// Create leader index:
|
||||
CreateIndexRequest createIndexRequest = new CreateIndexRequest("leader");
|
||||
createIndexRequest.settings(Collections.singletonMap("index.soft_deletes.enabled", true));
|
||||
CreateIndexResponse response = client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
|
||||
assertThat(response.isAcknowledged(), is(true));
|
||||
}
|
||||
String followIndex = "follower";
|
||||
// Follow index, so that it can be paused:
|
||||
{
|
||||
// TODO: Replace this with high level rest client code when put follow API is available:
|
||||
final Request request = new Request("PUT", "/" + followIndex + "/_ccr/follow");
|
||||
request.setJsonEntity("{\"remote_cluster\": \"local\", \"leader_index\": \"leader\"}");
|
||||
Response response = client().performRequest(request);
|
||||
assertThat(response.getStatusLine().getStatusCode(), equalTo(200));
|
||||
}
|
||||
|
||||
// tag::ccr-pause-follow-request
|
||||
PauseFollowRequest request = new PauseFollowRequest(followIndex); // <1>
|
||||
// end::ccr-pause-follow-request
|
||||
|
||||
// tag::ccr-pause-follow-execute
|
||||
AcknowledgedResponse response =
|
||||
client.ccr().pauseFollow(request, RequestOptions.DEFAULT);
|
||||
// end::ccr-pause-follow-execute
|
||||
|
||||
// tag::ccr-pause-follow-response
|
||||
boolean acknowledged = response.isAcknowledged(); // <1>
|
||||
// end::ccr-pause-follow-response
|
||||
|
||||
// tag::ccr-pause-follow-execute-listener
|
||||
ActionListener<AcknowledgedResponse> listener =
|
||||
new ActionListener<AcknowledgedResponse>() {
|
||||
@Override
|
||||
public void onResponse(AcknowledgedResponse response) {
|
||||
boolean acknowledged = response.isAcknowledged(); // <1>
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onFailure(Exception e) {
|
||||
// <2>
|
||||
}
|
||||
};
|
||||
// end::ccr-pause-follow-execute-listener
|
||||
|
||||
// Resume follow index, so that it can be paused again:
|
||||
{
|
||||
// TODO: Replace this with high level rest client code when resume follow API is available:
|
||||
final Request req = new Request("POST", "/" + followIndex + "/_ccr/resume_follow");
|
||||
req.setJsonEntity("{}");
|
||||
Response res = client().performRequest(req);
|
||||
assertThat(res.getStatusLine().getStatusCode(), equalTo(200));
|
||||
}
|
||||
|
||||
// Replace the empty listener by a blocking listener in test
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
listener = new LatchedActionListener<>(listener, latch);
|
||||
|
||||
// tag::ccr-pause-follow-execute-async
|
||||
client.ccr()
|
||||
.pauseFollowAsync(request, RequestOptions.DEFAULT, listener); // <1>
|
||||
// end::ccr-pause-follow-execute-async
|
||||
|
||||
assertTrue(latch.await(30L, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
static Map<String, Object> toMap(Response response) throws IOException {
|
||||
return XContentHelper.convertToMap(JsonXContent.jsonXContent, EntityUtils.toString(response.getEntity()), false);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,35 @@
|
|||
--
|
||||
:api: ccr-pause-follow
|
||||
:request: PauseFollowRequest
|
||||
:response: PauseFollowResponse
|
||||
--
|
||||
|
||||
[id="{upid}-{api}"]
|
||||
=== Pause Follow API
|
||||
|
||||
|
||||
[id="{upid}-{api}-request"]
|
||||
==== Request
|
||||
|
||||
The Pause Follow API allows you to pause following by follow index name.
|
||||
|
||||
["source","java",subs="attributes,callouts,macros"]
|
||||
--------------------------------------------------
|
||||
include-tagged::{doc-tests-file}[{api}-request]
|
||||
--------------------------------------------------
|
||||
<1> The name of follow index.
|
||||
|
||||
[id="{upid}-{api}-response"]
|
||||
==== Response
|
||||
|
||||
The returned +{response}+ indicates if the pause follow request was received.
|
||||
|
||||
["source","java",subs="attributes,callouts,macros"]
|
||||
--------------------------------------------------
|
||||
include-tagged::{doc-tests-file}[{api}-response]
|
||||
--------------------------------------------------
|
||||
<1> Whether or not the pause follow was acknowledge.
|
||||
|
||||
include::../execution.asciidoc[]
|
||||
|
||||
|
||||
|
|
@ -397,3 +397,14 @@ don't leak into the rest of the documentation.
|
|||
:doc-tests-file!:
|
||||
:upid!:
|
||||
--
|
||||
|
||||
== CCR APIs
|
||||
|
||||
:upid: {mainid}-ccr
|
||||
:doc-tests-file: {doc-tests}/CCRDocumentationIT.java
|
||||
|
||||
The Java High Level REST Client supports the following CCR APIs:
|
||||
|
||||
* <<{upid}-ccr-pause-follow>>
|
||||
|
||||
include::ccr/pause_follow.asciidoc[]
|
||||
|
|
|
|||
Loading…
Reference in New Issue