Skip to content

Commit

Permalink
Add stop rollup job support to HL REST Client (#34702)
Browse files Browse the repository at this point in the history
This change adds support for stoping a rollup job to the High Level REST Client.

Relates to #29827
  • Loading branch information
Christoph Büscher committed Nov 14, 2018
1 parent c346a0f commit 603d1a4
Show file tree
Hide file tree
Showing 12 changed files with 349 additions and 20 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,14 @@
import org.elasticsearch.client.rollup.GetRollupJobResponse;
import org.elasticsearch.client.rollup.GetRollupCapsRequest;
import org.elasticsearch.client.rollup.GetRollupCapsResponse;
import org.elasticsearch.client.rollup.GetRollupJobRequest;
import org.elasticsearch.client.rollup.GetRollupJobResponse;
import org.elasticsearch.client.rollup.PutRollupJobRequest;
import org.elasticsearch.client.rollup.PutRollupJobResponse;
import org.elasticsearch.client.rollup.StartRollupJobRequest;
import org.elasticsearch.client.rollup.StartRollupJobResponse;
import org.elasticsearch.client.rollup.StopRollupJobRequest;
import org.elasticsearch.client.rollup.StopRollupJobResponse;

import java.io.IOException;
import java.util.Collections;
Expand Down Expand Up @@ -118,6 +122,40 @@ public void startRollupJobAsync(StartRollupJobRequest request, RequestOptions op
listener, Collections.emptySet());
}

/**
* Stop a rollup job
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/rollup-stop-job.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 StopRollupJobResponse stopRollupJob(StopRollupJobRequest request, RequestOptions options) throws IOException {
return restHighLevelClient.performRequestAndParseEntity(request,
RollupRequestConverters::stopJob,
options,
StopRollupJobResponse::fromXContent,
Collections.emptySet());
}

/**
* Asynchronously stop a rollup job
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/rollup-stop-job.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
* @param listener the listener to be notified upon request completion
*/
public void stopRollupJobAsync(StopRollupJobRequest request, RequestOptions options,
ActionListener<StopRollupJobResponse> listener) {
restHighLevelClient.performRequestAsyncAndParseEntity(request,
RollupRequestConverters::stopJob,
options,
StopRollupJobResponse::fromXContent,
listener, Collections.emptySet());
}

/**
* Delete a rollup job from the cluster
* See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/rollup-delete-job.html">
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.elasticsearch.client.rollup.GetRollupJobRequest;
import org.elasticsearch.client.rollup.PutRollupJobRequest;
import org.elasticsearch.client.rollup.StartRollupJobRequest;
import org.elasticsearch.client.rollup.StopRollupJobRequest;

import java.io.IOException;

Expand Down Expand Up @@ -55,8 +56,16 @@ static Request startJob(final StartRollupJobRequest startRollupJobRequest) throw
.addPathPart(startRollupJobRequest.getJobId())
.addPathPartAsIs("_start")
.build();
Request request = new Request(HttpPost.METHOD_NAME, endpoint);
return request;
return new Request(HttpPost.METHOD_NAME, endpoint);
}

static Request stopJob(final StopRollupJobRequest stopRollupJobRequest) throws IOException {
String endpoint = new RequestConverters.EndpointBuilder()
.addPathPartAsIs("_xpack", "rollup", "job")
.addPathPart(stopRollupJobRequest.getJobId())
.addPathPartAsIs("_stop")
.build();
return new Request(HttpPost.METHOD_NAME, endpoint);
}

static Request getJob(final GetRollupJobRequest getRollupJobRequest) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ public class StartRollupJobResponse extends AcknowledgedResponse {
private static final String PARSE_FIELD_NAME = "started";

private static final ConstructingObjectParser<StartRollupJobResponse, Void> PARSER = AcknowledgedResponse
.generateParser("delete_rollup_job_response", StartRollupJobResponse::new, PARSE_FIELD_NAME);
.generateParser("start_rollup_job_response", StartRollupJobResponse::new, PARSE_FIELD_NAME);

public StartRollupJobResponse(boolean acknowledged) {
super(acknowledged);
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
/*
* 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.rollup;

import org.elasticsearch.client.Validatable;

import java.util.Objects;

public class StopRollupJobRequest implements Validatable {

private final String jobId;

public StopRollupJobRequest(final String jobId) {
this.jobId = Objects.requireNonNull(jobId, "id parameter must not be null");
}

public String getJobId() {
return jobId;
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
final StopRollupJobRequest that = (StopRollupJobRequest) o;
return Objects.equals(jobId, that.jobId);
}

@Override
public int hashCode() {
return Objects.hash(jobId);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
/*
* 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.rollup;

import org.elasticsearch.client.core.AcknowledgedResponse;
import org.elasticsearch.common.xcontent.ConstructingObjectParser;
import org.elasticsearch.common.xcontent.XContentParser;

import java.io.IOException;

public class StopRollupJobResponse extends AcknowledgedResponse {

private static final String PARSE_FIELD_NAME = "stopped";

private static final ConstructingObjectParser<StopRollupJobResponse, Void> PARSER = AcknowledgedResponse
.generateParser("stop_rollup_job_response", StopRollupJobResponse::new, PARSE_FIELD_NAME);

public StopRollupJobResponse(boolean acknowledged) {
super(acknowledged);
}

public static StopRollupJobResponse fromXContent(final XContentParser parser) throws IOException {
return PARSER.parse(parser, null);
}

@Override
protected String getFieldName() {
return PARSE_FIELD_NAME;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -41,10 +41,12 @@
import org.elasticsearch.client.rollup.GetRollupJobResponse.JobWrapper;
import org.elasticsearch.client.rollup.PutRollupJobRequest;
import org.elasticsearch.client.rollup.PutRollupJobResponse;
import org.elasticsearch.client.rollup.StartRollupJobRequest;
import org.elasticsearch.client.rollup.StartRollupJobResponse;
import org.elasticsearch.client.rollup.RollableIndexCaps;
import org.elasticsearch.client.rollup.RollupJobCaps;
import org.elasticsearch.client.rollup.StartRollupJobRequest;
import org.elasticsearch.client.rollup.StartRollupJobResponse;
import org.elasticsearch.client.rollup.StopRollupJobRequest;
import org.elasticsearch.client.rollup.StopRollupJobResponse;
import org.elasticsearch.client.rollup.job.config.DateHistogramGroupConfig;
import org.elasticsearch.client.rollup.job.config.GroupConfig;
import org.elasticsearch.client.rollup.job.config.MetricConfig;
Expand Down Expand Up @@ -230,6 +232,11 @@ public void testPutStartAndGetRollupJob() throws Exception {
assertThat(job.getStatus().getState(), either(equalTo(IndexerState.STARTED)).or(equalTo(IndexerState.INDEXING)));
assertThat(job.getStatus().getCurrentPosition(), hasKey("date.date_histogram"));
assertEquals(true, job.getStatus().getUpgradedDocumentId());

// stop the job
StopRollupJobRequest stopRequest = new StopRollupJobRequest(id);
StopRollupJobResponse stopResponse = execute(stopRequest, rollupClient::stopRollupJob, rollupClient::stopRollupJobAsync);
assertTrue(stopResponse.isAcknowledged());
}

public void testGetMissingRollupJob() throws Exception {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.elasticsearch.client.rollup.GetRollupJobRequest;
import org.elasticsearch.client.rollup.PutRollupJobRequest;
import org.elasticsearch.client.rollup.StartRollupJobRequest;
import org.elasticsearch.client.rollup.StopRollupJobRequest;
import org.elasticsearch.client.rollup.job.config.RollupJobConfig;
import org.elasticsearch.client.rollup.job.config.RollupJobConfigTests;
import org.elasticsearch.test.ESTestCase;
Expand Down Expand Up @@ -61,6 +62,18 @@ public void testStartJob() throws IOException {
assertThat(request.getEntity(), nullValue());
}

public void testStopJob() throws IOException {
String jobId = randomAlphaOfLength(5);

StopRollupJobRequest stopJob = new StopRollupJobRequest(jobId);

Request request = RollupRequestConverters.stopJob(stopJob);
assertThat(request.getEndpoint(), equalTo("/_xpack/rollup/job/" + jobId + "/_stop"));
assertThat(HttpPost.METHOD_NAME, equalTo(request.getMethod()));
assertThat(request.getParameters().keySet(), empty());
assertNull(request.getEntity());
}

public void testGetJob() {
boolean getAll = randomBoolean();
String job = getAll ? "_all" : RequestConvertersTests.randomIndicesNames(1, 1)[0];
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,7 @@
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.support.WriteRequest;
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.RollupClient;
import org.elasticsearch.client.rollup.DeleteRollupJobRequest;
Expand All @@ -51,6 +49,8 @@
import org.elasticsearch.client.rollup.RollupJobCaps;
import org.elasticsearch.client.rollup.StartRollupJobRequest;
import org.elasticsearch.client.rollup.StartRollupJobResponse;
import org.elasticsearch.client.rollup.StopRollupJobRequest;
import org.elasticsearch.client.rollup.StopRollupJobResponse;
import org.elasticsearch.client.rollup.job.config.DateHistogramGroupConfig;
import org.elasticsearch.client.rollup.job.config.GroupConfig;
import org.elasticsearch.client.rollup.job.config.HistogramGroupConfig;
Expand Down Expand Up @@ -237,59 +237,96 @@ public void onFailure(Exception e) {
assertTrue(latch.await(30L, TimeUnit.SECONDS));
}


@SuppressWarnings("unused")
public void testStartRollupJob() throws Exception {
testCreateRollupJob();
RestHighLevelClient client = highLevelClient();

String id = "job_1";
// tag::rollup-start-job-request
StartRollupJobRequest request = new StartRollupJobRequest(id); // <1>
// end::rollup-start-job-request


try {
// tag::rollup-start-job-execute
RollupClient rc = client.rollup();
StartRollupJobResponse response = rc.startRollupJob(request, RequestOptions.DEFAULT);
// end::rollup-start-job-execute

// tag::rollup-start-job-response
response.isAcknowledged(); // <1>
// end::rollup-start-job-response
} catch (Exception e) {
// Swallow any exception, this test does not test actually cancelling.
}

// tag::rollup-start-job-execute-listener
ActionListener<StartRollupJobResponse> listener = new ActionListener<StartRollupJobResponse>() {
@Override
public void onResponse(StartRollupJobResponse response) {
// <1>
}

@Override
public void onFailure(Exception e) {
// <2>
}
};
// end::rollup-start-job-execute-listener

final CountDownLatch latch = new CountDownLatch(1);
listener = new LatchedActionListener<>(listener, latch);

// tag::rollup-start-job-execute-async
RollupClient rc = client.rollup();
rc.startRollupJobAsync(request, RequestOptions.DEFAULT, listener); // <1>
// end::rollup-start-job-execute-async

assertTrue(latch.await(30L, TimeUnit.SECONDS));

// stop job so it can correctly be deleted by the test teardown
// TODO Replace this with the Rollup Stop Job API
Response stoptResponse = client().performRequest(new Request("POST", "/_xpack/rollup/job/" + id + "/_stop"));
assertEquals(RestStatus.OK.getStatus(), stoptResponse.getStatusLine().getStatusCode());
rc.stopRollupJob(new StopRollupJobRequest(id), RequestOptions.DEFAULT);
}

@SuppressWarnings("unused")
public void testStopRollupJob() throws Exception {
testCreateRollupJob();
RestHighLevelClient client = highLevelClient();

String id = "job_1";
// tag::rollup-stop-job-request
StopRollupJobRequest request = new StopRollupJobRequest(id); // <1>
// end::rollup-stop-job-request


try {
// tag::rollup-stop-job-execute
RollupClient rc = client.rollup();
StopRollupJobResponse response = rc.stopRollupJob(request, RequestOptions.DEFAULT);
// end::rollup-stop-job-execute

// tag::rollup-stop-job-response
response.isAcknowledged(); // <1>
// end::rollup-stop-job-response
} catch (Exception e) {
// Swallow any exception, this test does not test actually cancelling.
}

// tag::rollup-stop-job-execute-listener
ActionListener<StopRollupJobResponse> listener = new ActionListener<StopRollupJobResponse>() {
@Override
public void onResponse(StopRollupJobResponse response) {
// <1>
}

@Override
public void onFailure(Exception e) {
// <2>
}
};
// end::rollup-stop-job-execute-listener

final CountDownLatch latch = new CountDownLatch(1);
listener = new LatchedActionListener<>(listener, latch);

// tag::rollup-stop-job-execute-async
RollupClient rc = client.rollup();
rc.stopRollupJobAsync(request, RequestOptions.DEFAULT, listener); // <1>
// end::rollup-stop-job-execute-async

assertTrue(latch.await(30L, TimeUnit.SECONDS));
}

@SuppressWarnings("unused")
Expand Down

0 comments on commit 603d1a4

Please sign in to comment.