diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/Ccr.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/Ccr.java
index 0de5dd3c1a45e..8d677866e3219 100644
--- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/Ccr.java
+++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/Ccr.java
@@ -162,7 +162,7 @@ public Collection<Object> createComponents(
ccrLicenseChecker,
restoreSourceService,
new CcrRepositoryManager(settings, clusterService, client),
- new AutoFollowCoordinator(client, clusterService, ccrLicenseChecker, threadPool::relativeTimeInMillis)
+ new AutoFollowCoordinator(settings, client, clusterService, ccrLicenseChecker, threadPool::relativeTimeInMillis)
);
}
diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/CcrSettings.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/CcrSettings.java
index 544a45792e070..d7495dec8c2cf 100644
--- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/CcrSettings.java
+++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/CcrSettings.java
@@ -7,6 +7,7 @@
import org.elasticsearch.common.settings.Setting;
import org.elasticsearch.common.settings.Setting.Property;
+import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.xpack.core.XPackSettings;
import java.util.Arrays;
@@ -28,6 +29,12 @@ private CcrSettings() {
public static final Setting<Boolean> CCR_FOLLOWING_INDEX_SETTING =
Setting.boolSetting("index.xpack.ccr.following_index", false, Property.IndexScope, Property.InternalIndex);
+ /**
+ * Dynamic node setting for specifying the wait_for_timeout that the auto follow coordinator should be using.
+ */
+ public static final Setting<TimeValue> CCR_AUTO_FOLLOW_WAIT_FOR_METADATA_TIMEOUT = Setting.timeSetting(
+ "ccr.auto_follow.wait_for_metadata_timeout", TimeValue.timeValueSeconds(60), Property.NodeScope, Property.Dynamic);
+
/**
* The settings defined by CCR.
*
@@ -36,7 +43,8 @@ private CcrSettings() {
static List<Setting<?>> getSettings() {
return Arrays.asList(
XPackSettings.CCR_ENABLED_SETTING,
- CCR_FOLLOWING_INDEX_SETTING);
+ CCR_FOLLOWING_INDEX_SETTING,
+ CCR_AUTO_FOLLOW_WAIT_FOR_METADATA_TIMEOUT);
}
}
diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinator.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinator.java
index 9043b379b5b33..5cc5920cd213a 100644
--- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinator.java
+++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinator.java
@@ -27,12 +27,14 @@
import org.elasticsearch.common.collect.CopyOnWriteHashMap;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.settings.Settings;
+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.xpack.ccr.CcrLicenseChecker;
+import org.elasticsearch.xpack.ccr.CcrSettings;
import org.elasticsearch.xpack.core.ccr.AutoFollowMetadata;
import org.elasticsearch.xpack.core.ccr.AutoFollowMetadata.AutoFollowPattern;
import org.elasticsearch.xpack.core.ccr.AutoFollowStats;
@@ -73,6 +75,7 @@ public class AutoFollowCoordinator implements ClusterStateListener {
private final CcrLicenseChecker ccrLicenseChecker;
private final LongSupplier relativeMillisTimeProvider;
+ private volatile TimeValue waitForMetadataTimeOut;
private volatile Map<String, AutoFollower> autoFollowers = Collections.emptyMap();
// The following fields are read and updated under a lock:
@@ -82,6 +85,7 @@ public class AutoFollowCoordinator implements ClusterStateListener {
private final LinkedHashMap<String, ElasticsearchException> recentAutoFollowErrors;
public AutoFollowCoordinator(
+ Settings settings,
Client client,
ClusterService clusterService,
CcrLicenseChecker ccrLicenseChecker,
@@ -98,6 +102,15 @@ protected boolean removeEldestEntry(final Map.Entry<String, ElasticsearchExcepti
return size() > MAX_AUTO_FOLLOW_ERRORS;
}
};
+
+ Consumer<TimeValue> updater = newWaitForTimeOut -> {
+ if (newWaitForTimeOut.equals(waitForMetadataTimeOut) == false) {
+ LOGGER.info("changing wait_for_metadata_timeout from [{}] to [{}]", waitForMetadataTimeOut, newWaitForTimeOut);
+ waitForMetadataTimeOut = newWaitForTimeOut;
+ }
+ };
+ clusterService.getClusterSettings().addSettingsUpdateConsumer(CcrSettings.CCR_AUTO_FOLLOW_WAIT_FOR_METADATA_TIMEOUT, updater);
+ waitForMetadataTimeOut = CcrSettings.CCR_AUTO_FOLLOW_WAIT_FOR_METADATA_TIMEOUT.get(settings);
}
public synchronized AutoFollowStats getStats() {
@@ -181,6 +194,7 @@ void getRemoteClusterState(final String remoteCluster,
request.metaData(true);
request.routingTable(true);
request.waitForMetaDataVersion(metadataVersion);
+ request.waitForTimeout(waitForMetadataTimeOut);
// TODO: set non-compliant status on auto-follow coordination that can be viewed via a stats API
ccrLicenseChecker.checkRemoteClusterLicenseAndFetchClusterState(
client,
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrIntegTestCase.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrIntegTestCase.java
index 1d705934cce35..01e51ea94f255 100644
--- a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrIntegTestCase.java
+++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrIntegTestCase.java
@@ -65,6 +65,7 @@
import org.elasticsearch.test.TestCluster;
import org.elasticsearch.test.discovery.TestZenDiscovery;
import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.ccr.CcrSettings;
import org.elasticsearch.xpack.ccr.LocalStateCcr;
import org.elasticsearch.xpack.ccr.index.engine.FollowingEngine;
import org.elasticsearch.xpack.core.XPackSettings;
@@ -199,6 +200,8 @@ private NodeConfigurationSource createNodeConfigurationSource(String leaderSeedA
builder.put(XPackSettings.MACHINE_LEARNING_ENABLED.getKey(), false);
builder.put(XPackSettings.LOGSTASH_ENABLED.getKey(), false);
builder.put(LicenseService.SELF_GENERATED_LICENSE_TYPE.getKey(), "trial");
+ // Let cluster state api return quickly in order to speed up auto follow tests:
+ builder.put(CcrSettings.CCR_AUTO_FOLLOW_WAIT_FOR_METADATA_TIMEOUT.getKey(), TimeValue.timeValueMillis(100));
if (leaderSeedAddress != null) {
builder.put("cluster.remote.leader_cluster.seeds", leaderSeedAddress);
}
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrSingleNodeTestCase.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrSingleNodeTestCase.java
index 2b1c8a8ef2edc..2fb1f868dd7fb 100644
--- a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrSingleNodeTestCase.java
+++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrSingleNodeTestCase.java
@@ -15,6 +15,7 @@
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.test.ESSingleNodeTestCase;
import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.ccr.CcrSettings;
import org.elasticsearch.xpack.ccr.LocalStateCcr;
import org.elasticsearch.xpack.core.XPackSettings;
import org.elasticsearch.xpack.core.ccr.ShardFollowNodeTaskStatus;
@@ -42,6 +43,8 @@ protected Settings nodeSettings() {
builder.put(XPackSettings.MACHINE_LEARNING_ENABLED.getKey(), false);
builder.put(XPackSettings.LOGSTASH_ENABLED.getKey(), false);
builder.put(LicenseService.SELF_GENERATED_LICENSE_TYPE.getKey(), "trial");
+ // Let cluster state api return quickly in order to speed up auto follow tests:
+ builder.put(CcrSettings.CCR_AUTO_FOLLOW_WAIT_FOR_METADATA_TIMEOUT.getKey(), TimeValue.timeValueMillis(100));
return builder.build();
}
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinatorTests.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinatorTests.java
index 2e684e2a48600..3acdde52a443e 100644
--- a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinatorTests.java
+++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinatorTests.java
@@ -19,11 +19,13 @@
import org.elasticsearch.cluster.routing.TestShardRouting;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.settings.ClusterSettings;
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.CcrSettings;
import org.elasticsearch.xpack.ccr.action.AutoFollowCoordinator.AutoFollower;
import org.elasticsearch.xpack.core.ccr.AutoFollowMetadata;
import org.elasticsearch.xpack.core.ccr.AutoFollowMetadata.AutoFollowPattern;
@@ -533,8 +535,9 @@ public void testGetFollowerIndexName() {
public void testStats() {
AutoFollowCoordinator autoFollowCoordinator = new AutoFollowCoordinator(
+ Settings.EMPTY,
null,
- mock(ClusterService.class),
+ mockClusterService(),
new CcrLicenseChecker(() -> true, () -> false),
() -> 1L);
@@ -589,7 +592,7 @@ public void testStats() {
}
public void testUpdateAutoFollowers() {
- ClusterService clusterService = mock(ClusterService.class);
+ ClusterService clusterService = mockClusterService();
// Return a cluster state with no patterns so that the auto followers never really execute:
ClusterState followerState = ClusterState.builder(new ClusterName("remote"))
.metaData(MetaData.builder().putCustom(AutoFollowMetadata.TYPE,
@@ -597,6 +600,7 @@ public void testUpdateAutoFollowers() {
.build();
when(clusterService.state()).thenReturn(followerState);
AutoFollowCoordinator autoFollowCoordinator = new AutoFollowCoordinator(
+ Settings.EMPTY,
null,
clusterService,
new CcrLicenseChecker(() -> true, () -> false),
@@ -651,8 +655,9 @@ public void testUpdateAutoFollowers() {
public void testUpdateAutoFollowersNoPatterns() {
AutoFollowCoordinator autoFollowCoordinator = new AutoFollowCoordinator(
+ Settings.EMPTY,
null,
- mock(ClusterService.class),
+ mockClusterService(),
new CcrLicenseChecker(() -> true, () -> false),
() -> 1L);
ClusterState clusterState = ClusterState.builder(new ClusterName("remote"))
@@ -665,8 +670,9 @@ public void testUpdateAutoFollowersNoPatterns() {
public void testUpdateAutoFollowersNoAutoFollowMetadata() {
AutoFollowCoordinator autoFollowCoordinator = new AutoFollowCoordinator(
+ Settings.EMPTY,
null,
- mock(ClusterService.class),
+ mockClusterService(),
new CcrLicenseChecker(() -> true, () -> false),
() -> 1L);
ClusterState clusterState = ClusterState.builder(new ClusterName("remote")).build();
@@ -918,4 +924,12 @@ private static Supplier<ClusterState> localClusterStateSupplier(ClusterState...
};
}
+ private ClusterService mockClusterService() {
+ ClusterService clusterService = mock(ClusterService.class);
+ ClusterSettings clusterSettings =
+ new ClusterSettings(Settings.EMPTY, Collections.singleton(CcrSettings.CCR_AUTO_FOLLOW_WAIT_FOR_METADATA_TIMEOUT));
+ when(clusterService.getClusterSettings()).thenReturn(clusterSettings);
+ return clusterService;
+ }
+
}