|
24 | 24 | import org.elasticsearch.action.LatchedActionListener;
|
25 | 25 | import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
|
26 | 26 | import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsResponse;
|
| 27 | +import org.elasticsearch.action.admin.indices.close.CloseIndexRequest; |
27 | 28 | import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
|
28 | 29 | import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
|
29 | 30 | import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
|
|
35 | 36 | import org.elasticsearch.client.ccr.PauseFollowRequest;
|
36 | 37 | import org.elasticsearch.client.ccr.PutFollowRequest;
|
37 | 38 | import org.elasticsearch.client.ccr.PutFollowResponse;
|
| 39 | +import org.elasticsearch.client.ccr.UnfollowRequest; |
38 | 40 | import org.elasticsearch.client.core.AcknowledgedResponse;
|
39 | 41 | import org.elasticsearch.common.xcontent.XContentHelper;
|
40 | 42 | import org.elasticsearch.common.xcontent.json.JsonXContent;
|
@@ -217,6 +219,91 @@ public void onFailure(Exception e) {
|
217 | 219 | assertTrue(latch.await(30L, TimeUnit.SECONDS));
|
218 | 220 | }
|
219 | 221 |
|
| 222 | + public void testUnfollow() throws Exception { |
| 223 | + RestHighLevelClient client = highLevelClient(); |
| 224 | + { |
| 225 | + // Create leader index: |
| 226 | + CreateIndexRequest createIndexRequest = new CreateIndexRequest("leader"); |
| 227 | + createIndexRequest.settings(Collections.singletonMap("index.soft_deletes.enabled", true)); |
| 228 | + CreateIndexResponse response = client.indices().create(createIndexRequest, RequestOptions.DEFAULT); |
| 229 | + assertThat(response.isAcknowledged(), is(true)); |
| 230 | + } |
| 231 | + String followIndex = "follower"; |
| 232 | + // Follow index, pause and close, so that it can be unfollowed: |
| 233 | + { |
| 234 | + PutFollowRequest putFollowRequest = new PutFollowRequest("local", "leader", followIndex); |
| 235 | + PutFollowResponse putFollowResponse = client.ccr().putFollow(putFollowRequest, RequestOptions.DEFAULT); |
| 236 | + assertThat(putFollowResponse.isFollowIndexCreated(), is(true)); |
| 237 | + assertThat(putFollowResponse.isFollowIndexShardsAcked(), is(true)); |
| 238 | + assertThat(putFollowResponse.isIndexFollowingStarted(), is(true)); |
| 239 | + |
| 240 | + PauseFollowRequest pauseFollowRequest = new PauseFollowRequest(followIndex); |
| 241 | + AcknowledgedResponse unfollowResponse = client.ccr().pauseFollow(pauseFollowRequest, RequestOptions.DEFAULT); |
| 242 | + assertThat(unfollowResponse.isAcknowledged(), is(true)); |
| 243 | + |
| 244 | + CloseIndexRequest closeIndexRequest = new CloseIndexRequest(followIndex); |
| 245 | + assertThat(client.indices().close(closeIndexRequest, RequestOptions.DEFAULT).isAcknowledged(), is(true)); |
| 246 | + } |
| 247 | + |
| 248 | + // tag::ccr-unfollow-request |
| 249 | + UnfollowRequest request = new UnfollowRequest(followIndex); // <1> |
| 250 | + // end::ccr-unfollow-request |
| 251 | + |
| 252 | + // tag::ccr-unfollow-execute |
| 253 | + AcknowledgedResponse response = |
| 254 | + client.ccr().unfollow(request, RequestOptions.DEFAULT); |
| 255 | + // end::ccr-unfollow-execute |
| 256 | + |
| 257 | + // tag::ccr-unfollow-response |
| 258 | + boolean acknowledged = response.isAcknowledged(); // <1> |
| 259 | + // end::ccr-unfollow-response |
| 260 | + |
| 261 | + // Delete, put follow index, pause and close, so that it can be unfollowed again: |
| 262 | + { |
| 263 | + DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(followIndex); |
| 264 | + assertThat(client.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT).isAcknowledged(), is(true)); |
| 265 | + |
| 266 | + PutFollowRequest putFollowRequest = new PutFollowRequest("local", "leader", followIndex); |
| 267 | + PutFollowResponse putFollowResponse = client.ccr().putFollow(putFollowRequest, RequestOptions.DEFAULT); |
| 268 | + assertThat(putFollowResponse.isFollowIndexCreated(), is(true)); |
| 269 | + assertThat(putFollowResponse.isFollowIndexShardsAcked(), is(true)); |
| 270 | + assertThat(putFollowResponse.isIndexFollowingStarted(), is(true)); |
| 271 | + |
| 272 | + PauseFollowRequest pauseFollowRequest = new PauseFollowRequest(followIndex); |
| 273 | + AcknowledgedResponse unfollowResponse = client.ccr().pauseFollow(pauseFollowRequest, RequestOptions.DEFAULT); |
| 274 | + assertThat(unfollowResponse.isAcknowledged(), is(true)); |
| 275 | + |
| 276 | + CloseIndexRequest closeIndexRequest = new CloseIndexRequest(followIndex); |
| 277 | + assertThat(client.indices().close(closeIndexRequest, RequestOptions.DEFAULT).isAcknowledged(), is(true)); |
| 278 | + } |
| 279 | + |
| 280 | + // tag::ccr-unfollow-execute-listener |
| 281 | + ActionListener<AcknowledgedResponse> listener = |
| 282 | + new ActionListener<AcknowledgedResponse>() { |
| 283 | + @Override |
| 284 | + public void onResponse(AcknowledgedResponse response) { |
| 285 | + boolean acknowledged = response.isAcknowledged(); // <1> |
| 286 | + } |
| 287 | + |
| 288 | + @Override |
| 289 | + public void onFailure(Exception e) { |
| 290 | + // <2> |
| 291 | + } |
| 292 | + }; |
| 293 | + // end::ccr-unfollow-execute-listener |
| 294 | + |
| 295 | + // Replace the empty listener by a blocking listener in test |
| 296 | + final CountDownLatch latch = new CountDownLatch(1); |
| 297 | + listener = new LatchedActionListener<>(listener, latch); |
| 298 | + |
| 299 | + // tag::ccr-unfollow-execute-async |
| 300 | + client.ccr() |
| 301 | + .unfollowAsync(request, RequestOptions.DEFAULT, listener); // <1> |
| 302 | + // end::ccr-unfollow-execute-async |
| 303 | + |
| 304 | + assertTrue(latch.await(30L, TimeUnit.SECONDS)); |
| 305 | + } |
| 306 | + |
220 | 307 | static Map<String, Object> toMap(Response response) throws IOException {
|
221 | 308 | return XContentHelper.convertToMap(JsonXContent.jsonXContent, EntityUtils.toString(response.getEntity()), false);
|
222 | 309 | }
|
|
0 commit comments