From b59607f69e77bd7fcac93bda98c932104556a5b3 Mon Sep 17 00:00:00 2001 From: Martijn van Groningen Date: Fri, 15 Feb 2019 16:10:27 +0100 Subject: [PATCH] Test bi-directional index following during a rolling upgrade. Follow index in follow cluster that follows an index in the leader cluster and another follow index in the leader index that follows that index in the follow cluster. During the upgrade index following is paused and after the upgrade index following is resumed and then verified index following works as expected. Relates to #38037 --- .../upgrades/CcrRollingUpgradeIT.java | 67 ++++++++++++++++++- 1 file changed, 65 insertions(+), 2 deletions(-) diff --git a/x-pack/qa/rolling-upgrade-multi-cluster/src/test/java/org/elasticsearch/upgrades/CcrRollingUpgradeIT.java b/x-pack/qa/rolling-upgrade-multi-cluster/src/test/java/org/elasticsearch/upgrades/CcrRollingUpgradeIT.java index b142d5bafaa48..556fb92f77bc7 100644 --- a/x-pack/qa/rolling-upgrade-multi-cluster/src/test/java/org/elasticsearch/upgrades/CcrRollingUpgradeIT.java +++ b/x-pack/qa/rolling-upgrade-multi-cluster/src/test/java/org/elasticsearch/upgrades/CcrRollingUpgradeIT.java @@ -22,7 +22,7 @@ public class CcrRollingUpgradeIT extends AbstractMultiClusterUpgradeTestCase { - public void testIndexFollowing() throws Exception { + public void testUniDirectionalIndexFollowing() throws Exception { logger.info("clusterName={}, upgradeState={}", clusterName, upgradeState); if (clusterName == ClusterName.LEADER) { @@ -230,6 +230,61 @@ public void testCannotFollowLeaderInUpgradedCluster() throws Exception { } } + public void testBiDirectionalIndexFollowing() throws Exception { + logger.info("clusterName={}, upgradeState={}", clusterName, upgradeState); + + if (clusterName == ClusterName.FOLLOWER) { + switch (upgradeState) { + case NONE: + createLeaderIndex(leaderClient(), "leader_index5"); + index(leaderClient(), "leader_index5", 128); + + followIndex(followerClient(), "leader", "leader_index5", "follower_index5"); + followIndex(leaderClient(), "follower", "follower_index5", "follower_index6"); + assertTotalHitCount("follower_index5", 128, followerClient()); + assertTotalHitCount("follower_index6", 128, leaderClient()); + + index(leaderClient(), "leader_index5", 128); + pauseIndexFollowing(followerClient(), "follower_index5"); + pauseIndexFollowing(leaderClient(), "follower_index6"); + break; + case ONE_THIRD: + index(leaderClient(), "leader_index5", 128); + break; + case TWO_THIRD: + index(leaderClient(), "leader_index5", 128); + break; + case ALL: + index(leaderClient(), "leader_index5", 128); + break; + default: + throw new AssertionError("unexpected upgrade_state [" + upgradeState + "]"); + } + } else if (clusterName == ClusterName.LEADER) { + switch (upgradeState) { + case NONE: + break; + case ONE_THIRD: + index(leaderClient(), "leader_index5", 128); + break; + case TWO_THIRD: + index(leaderClient(), "leader_index5", 128); + break; + case ALL: + resumeIndexFollowing(followerClient(), "follower_index5"); + resumeIndexFollowing(leaderClient(), "follower_index6"); + + assertTotalHitCount("follower_index5", 896, followerClient()); + assertTotalHitCount("follower_index6", 896, leaderClient()); + break; + default: + throw new AssertionError("unexpected upgrade_state [" + upgradeState + "]"); + } + } else { + throw new AssertionError("unexpected cluster_name [" + clusterName + "]"); + } + } + private static void createLeaderIndex(RestClient client, String indexName) throws IOException { Settings.Builder indexSettings = Settings.builder() .put("index.number_of_shards", 1) @@ -305,9 +360,17 @@ private static void verifyTotalHitCount(final String index, } private static void stopIndexFollowing(RestClient client, String followerIndex) throws IOException { - assertOK(client.performRequest(new Request("POST", "/" + followerIndex + "/_ccr/pause_follow"))); + pauseIndexFollowing(client, followerIndex); assertOK(client.performRequest(new Request("POST", "/" + followerIndex + "/_close"))); assertOK(client.performRequest(new Request("POST", "/" + followerIndex + "/_ccr/unfollow"))); } + private static void pauseIndexFollowing(RestClient client, String followerIndex) throws IOException { + assertOK(client.performRequest(new Request("POST", "/" + followerIndex + "/_ccr/pause_follow"))); + } + + private static void resumeIndexFollowing(RestClient client, String followerIndex) throws IOException { + assertOK(client.performRequest(new Request("POST", "/" + followerIndex + "/_ccr/resume_follow"))); + } + }