CCR: Do not follow if leader does not have soft-deletes (#34767)

We should not create a follower index and abort a follow request if the
leader does not have soft-deletes. Moreover, we also should not
auto-follow an index if it does not have soft-deletes.

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinator.java

@@ -26,6 +26,7 @@ import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
 import org.elasticsearch.common.util.concurrent.CountDown;
 import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.license.LicenseUtils;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.ccr.CcrLicenseChecker;
@@ -370,7 +371,9 @@ public class AutoFollowCoordinator implements ClusterStateApplier {
                         // has a leader index uuid custom metadata entry that matches with uuid of leaderIndexMetaData variable
                         // If so then handle it differently: not follow it, but just add an entry to
                         // AutoFollowMetadata#followedLeaderIndexUUIDs
-                        leaderIndicesToFollow.add(leaderIndexMetaData.getIndex());
+                        if (leaderIndexMetaData.getSettings().getAsBoolean(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), false)) {
+                            leaderIndicesToFollow.add(leaderIndexMetaData.getIndex());
+                        }
                     }
                 }
             }

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/TransportPutFollowAction.java

@@ -125,6 +125,10 @@ public final class TransportPutFollowAction
             listener.onFailure(new IllegalArgumentException("leader index [" + request.getLeaderIndex() + "] does not exist"));
             return;
         }
+        if (leaderIndexMetaData.getSettings().getAsBoolean(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), false) == false) {
+            listener.onFailure(
+                new IllegalArgumentException("leader index [" + request.getLeaderIndex() + "] does not have soft deletes enabled"));
+        }
 
         ActionListener<Boolean> handler = ActionListener.wrap(
                 result -> {

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/LocalIndexFollowingIT.java

@@ -67,6 +67,21 @@ public class LocalIndexFollowingIT extends CcrSingleNodeTestCase {
         });
     }
 
+    public void testDoNotCreateFollowerIfLeaderDoesNotHaveSoftDeletes() throws Exception {
+        final String leaderIndexSettings = getIndexSettings(2, 0,
+            singletonMap(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "false"));
+        assertAcked(client().admin().indices().prepareCreate("leader-index").setSource(leaderIndexSettings, XContentType.JSON));
+        ResumeFollowAction.Request followRequest = getResumeFollowRequest();
+        followRequest.setFollowerIndex("follower-index");
+        PutFollowAction.Request putFollowRequest = getPutFollowRequest();
+        putFollowRequest.setLeaderIndex("leader-index");
+        putFollowRequest.setFollowRequest(followRequest);
+        IllegalArgumentException error = expectThrows(IllegalArgumentException.class,
+            () -> client().execute(PutFollowAction.INSTANCE, putFollowRequest).actionGet());
+        assertThat(error.getMessage(), equalTo("leader index [leader-index] does not have soft deletes enabled"));
+        assertThat(client().admin().indices().prepareExists("follower-index").get().isExists(), equalTo(false));
+    }
+
     private String getIndexSettings(final int numberOfShards, final int numberOfReplicas,
                                     final Map<String, String> additionalIndexSettings) throws IOException {
         final String settings;

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinatorTests.java

@@ -15,6 +15,7 @@ import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.ccr.CcrLicenseChecker;
 import org.elasticsearch.xpack.ccr.action.AutoFollowCoordinator.AutoFollower;
@@ -50,7 +51,7 @@ public class AutoFollowCoordinatorTests extends ESTestCase {
 
         ClusterState leaderState = ClusterState.builder(new ClusterName("remote"))
             .metaData(MetaData.builder().put(IndexMetaData.builder("logs-20190101")
-                .settings(settings(Version.CURRENT))
+                .settings(settings(Version.CURRENT).put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true))
                 .numberOfShards(1)
                 .numberOfReplicas(0)))
             .build();
@@ -172,7 +173,7 @@ public class AutoFollowCoordinatorTests extends ESTestCase {
 
         ClusterState leaderState = ClusterState.builder(new ClusterName("remote"))
             .metaData(MetaData.builder().put(IndexMetaData.builder("logs-20190101")
-                .settings(settings(Version.CURRENT))
+                .settings(settings(Version.CURRENT).put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true))
                 .numberOfShards(1)
                 .numberOfReplicas(0)))
             .build();
@@ -235,7 +236,7 @@ public class AutoFollowCoordinatorTests extends ESTestCase {
 
         ClusterState leaderState = ClusterState.builder(new ClusterName("remote"))
             .metaData(MetaData.builder().put(IndexMetaData.builder("logs-20190101")
-                .settings(settings(Version.CURRENT))
+                .settings(settings(Version.CURRENT).put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true))
                 .numberOfShards(1)
                 .numberOfReplicas(0)))
             .build();
@@ -306,7 +307,8 @@ public class AutoFollowCoordinatorTests extends ESTestCase {
         for (int i = 0; i < 5; i++) {
             Settings.Builder builder = Settings.builder()
                 .put(IndexMetaData.SETTING_VERSION_CREATED, Version.CURRENT)
-                .put(IndexMetaData.SETTING_INDEX_UUID, "metrics-" + i);
+                .put(IndexMetaData.SETTING_INDEX_UUID, "metrics-" + i)
+                .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), i % 2 == 0);
             imdBuilder.put(IndexMetaData.builder("metrics-" + i)
                 .settings(builder)
                 .numberOfShards(1)
@@ -324,21 +326,17 @@ public class AutoFollowCoordinatorTests extends ESTestCase {
         List<Index> result = AutoFollower.getLeaderIndicesToFollow("remote", autoFollowPattern, leaderState, followerState,
             Collections.emptyList());
         result.sort(Comparator.comparing(Index::getName));
-        assertThat(result.size(), equalTo(5));
+        assertThat(result.size(), equalTo(3));
         assertThat(result.get(0).getName(), equalTo("metrics-0"));
-        assertThat(result.get(1).getName(), equalTo("metrics-1"));
+        assertThat(result.get(1).getName(), equalTo("metrics-2"));
-        assertThat(result.get(2).getName(), equalTo("metrics-2"));
+        assertThat(result.get(2).getName(), equalTo("metrics-4"));
-        assertThat(result.get(3).getName(), equalTo("metrics-3"));
-        assertThat(result.get(4).getName(), equalTo("metrics-4"));
 
         List<String> followedIndexUUIDs = Collections.singletonList(leaderState.metaData().index("metrics-2").getIndexUUID());
         result = AutoFollower.getLeaderIndicesToFollow("remote", autoFollowPattern, leaderState, followerState, followedIndexUUIDs);
         result.sort(Comparator.comparing(Index::getName));
-        assertThat(result.size(), equalTo(4));
+        assertThat(result.size(), equalTo(2));
         assertThat(result.get(0).getName(), equalTo("metrics-0"));
-        assertThat(result.get(1).getName(), equalTo("metrics-1"));
+        assertThat(result.get(1).getName(), equalTo("metrics-4"));
-        assertThat(result.get(2).getName(), equalTo("metrics-3"));
-        assertThat(result.get(3).getName(), equalTo("metrics-4"));
     }
 
     public void testGetFollowerIndexName() {