Introduce pluggable engine factory for CCR

This commit utilizes the pluggable engine factory feature in core to
introduce a pluggable engine factory for XDCR. For now this is only a
skeleton implementation to proof out the pluggable engine factory
concept. Future work will implement a genuine following engine for XDCR.

Relates #2655

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

@@ -7,10 +7,15 @@ package org.elasticsearch.xpack.ccr;
 
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.engine.EngineFactory;
+import org.elasticsearch.xpack.ccr.index.engine.FollowingEngineFactory;
 
 import java.util.List;
+import java.util.Optional;
 
 import static org.elasticsearch.xpack.ccr.CcrSettings.CCR_ENABLED_SETTING;
+import static org.elasticsearch.xpack.ccr.CcrSettings.CCR_FOLLOWING_INDEX_SETTING;
 
 /**
  * Container class for CCR functionality.
@@ -38,4 +43,18 @@ public final class Ccr {
         return CcrSettings.getSettings();
     }
 
+    /**
+     * The optional engine factory for CCR. This method inspects the index settings for the {@link CcrSettings#CCR_FOLLOWING_INDEX_SETTING}
+     * setting to determine whether or not the engine implementation should be a following engine.
+     *
+     * @return the optional engine factory
+     */
+    public Optional<EngineFactory> getEngineFactory(final IndexSettings indexSettings) {
+        if (CCR_FOLLOWING_INDEX_SETTING.get(indexSettings.getSettings())) {
+            return Optional.of(new FollowingEngineFactory());
+        } else {
+            return Optional.empty();
+        }
+    }
+
 }

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

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.ccr;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
@@ -26,8 +27,21 @@ final class CcrSettings {
      */
     static final Setting<Boolean> CCR_ENABLED_SETTING = Setting.boolSetting("xpack.ccr.enabled", true, Property.NodeScope);
 
+    /**
+     * Index setting for a following index.
+     */
+    static final Setting<Boolean> CCR_FOLLOWING_INDEX_SETTING =
+            Setting.boolSetting("index.xpack.ccr.following_index", false, Setting.Property.IndexScope);
+
+    /**
+     * The settings defined by CCR.
+     *
+     * @return the settings
+     */
     static List<Setting<?>> getSettings() {
-        return Collections.singletonList(CCR_ENABLED_SETTING);
+        return Arrays.asList(
+                CCR_ENABLED_SETTING,
+                CCR_FOLLOWING_INDEX_SETTING);
     }
 
 }

x-pack/plugin/src/main/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngineFactory.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.ccr.index.engine;
+
+import org.elasticsearch.index.engine.Engine;
+import org.elasticsearch.index.engine.EngineConfig;
+import org.elasticsearch.index.engine.EngineFactory;
+import org.elasticsearch.index.engine.InternalEngine;
+
+/**
+ * An engine factory for following engines.
+ */
+public class FollowingEngineFactory implements EngineFactory {
+
+    @Override
+    public Engine newReadWriteEngine(final EngineConfig config) {
+        // TODO: implement following engine
+        return new InternalEngine(config);
+    }
+
+}

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

@@ -0,0 +1,55 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.ccr;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.engine.EngineFactory;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.ccr.Ccr;
+import org.elasticsearch.xpack.ccr.index.engine.FollowingEngineFactory;
+
+import java.io.IOException;
+import java.util.Optional;
+
+import static org.hamcrest.Matchers.instanceOf;
+
+public class CcrTests extends ESTestCase {
+
+    public void testGetEngineFactory() throws IOException {
+        final Boolean[] values = new Boolean[] { true, false, null };
+        for (final Boolean value : values) {
+            final String indexName = "following-" + value;
+            final Index index = new Index(indexName, UUIDs.randomBase64UUID());
+            final Settings.Builder builder = Settings.builder()
+                    .put(IndexMetaData.SETTING_VERSION_CREATED, Version.CURRENT)
+                    .put(IndexMetaData.SETTING_INDEX_UUID, index.getUUID());
+            if (value != null) {
+                builder.put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), value);
+            }
+
+            final IndexMetaData indexMetaData = new IndexMetaData.Builder(index.getName())
+                    .settings(builder.build())
+                    .numberOfShards(1)
+                    .numberOfReplicas(0)
+                    .build();
+            final Ccr ccr = new Ccr(Settings.EMPTY);
+            final Optional<EngineFactory> engineFactory =
+                    ccr.getEngineFactory(new IndexSettings(indexMetaData, Settings.EMPTY));
+            if (value != null && value) {
+                assertTrue(engineFactory.isPresent());
+                assertThat(engineFactory.get(), instanceOf(FollowingEngineFactory.class));
+            } else {
+                assertFalse(engineFactory.isPresent());
+            }
+        }
+    }
+
+}