[CCR] Add validation for max_retry_delay (#33648)

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

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.ccr.action;
 
+import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.test.AbstractStreamableXContentTestCase;
@@ -12,6 +13,10 @@ import org.elasticsearch.xpack.core.ccr.action.FollowIndexAction;
 
 import java.io.IOException;
 
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
+
 public class FollowIndexRequestTests extends AbstractStreamableXContentTestCase<FollowIndexAction.Request> {
 
     @Override
@@ -39,4 +44,21 @@ public class FollowIndexRequestTests extends AbstractStreamableXContentTestCase<
             randomIntBetween(1, Integer.MAX_VALUE), randomNonNegativeLong(), randomIntBetween(1, Integer.MAX_VALUE),
             randomIntBetween(1, Integer.MAX_VALUE), TimeValue.timeValueMillis(500), TimeValue.timeValueMillis(500));
     }
+
+    public void testValidate() {
+        FollowIndexAction.Request request = new FollowIndexAction.Request("index1", "index2", null, null, null, null,
+            null, TimeValue.ZERO, null);
+        ActionRequestValidationException validationException = request.validate();
+        assertThat(validationException, notNullValue());
+        assertThat(validationException.getMessage(), containsString("[max_retry_delay] must be positive but was [0ms]"));
+
+        request = new FollowIndexAction.Request("index1", "index2", null, null, null, null, null, TimeValue.timeValueMinutes(10), null);
+        validationException = request.validate();
+        assertThat(validationException, notNullValue());
+        assertThat(validationException.getMessage(), containsString("[max_retry_delay] must be less than [5m] but was [10m]"));
+
+        request = new FollowIndexAction.Request("index1", "index2", null, null, null, null, null, TimeValue.timeValueMinutes(1), null);
+        validationException = request.validate();
+        assertThat(validationException, nullValue());
+    }
 }

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

@@ -169,7 +169,7 @@ public class AutoFollowMetadata extends AbstractNamedDiffable<MetaData.Custom> i
         public static final ParseField MAX_BATCH_SIZE_IN_BYTES = new ParseField("max_batch_size_in_bytes");
         public static final ParseField MAX_CONCURRENT_WRITE_BATCHES = new ParseField("max_concurrent_write_batches");
         public static final ParseField MAX_WRITE_BUFFER_SIZE = new ParseField("max_write_buffer_size");
-        public static final ParseField MAX_RETRY_DELAY = new ParseField("retry_timeout");
+        public static final ParseField MAX_RETRY_DELAY = new ParseField("max_retry_delay");
         public static final ParseField IDLE_SHARD_RETRY_DELAY = new ParseField("idle_shard_retry_delay");
 
         @SuppressWarnings("unchecked")

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

@@ -23,6 +23,8 @@ import org.elasticsearch.common.xcontent.XContentParser;
 import java.io.IOException;
 import java.util.Objects;
 
+import static org.elasticsearch.action.ValidateActions.addValidationError;
+
 public final class FollowIndexAction extends Action<AcknowledgedResponse> {
 
     public static final FollowIndexAction INSTANCE = new FollowIndexAction();
@@ -33,8 +35,9 @@ public final class FollowIndexAction extends Action<AcknowledgedResponse> {
     public static final int DEFAULT_MAX_CONCURRENT_READ_BATCHES = 1;
     public static final int DEFAULT_MAX_CONCURRENT_WRITE_BATCHES = 1;
     public static final long DEFAULT_MAX_BATCH_SIZE_IN_BYTES = Long.MAX_VALUE;
-    public static final TimeValue DEFAULT_RETRY_TIMEOUT = new TimeValue(500);
-    public static final TimeValue DEFAULT_IDLE_SHARD_RETRY_DELAY = TimeValue.timeValueSeconds(10);
+    static final TimeValue DEFAULT_MAX_RETRY_DELAY = new TimeValue(500);
+    static final TimeValue DEFAULT_IDLE_SHARD_RETRY_DELAY = TimeValue.timeValueSeconds(10);
+    static final TimeValue MAX_RETRY_DELAY = TimeValue.timeValueMinutes(5);
 
     private FollowIndexAction() {
         super(NAME);
@@ -54,7 +57,7 @@ public final class FollowIndexAction extends Action<AcknowledgedResponse> {
         private static final ParseField MAX_BATCH_SIZE_IN_BYTES = new ParseField("max_batch_size_in_bytes");
         private static final ParseField MAX_CONCURRENT_WRITE_BATCHES = new ParseField("max_concurrent_write_batches");
         private static final ParseField MAX_WRITE_BUFFER_SIZE = new ParseField("max_write_buffer_size");
-        private static final ParseField MAX_RETRY_DELAY = new ParseField("max_retry_delay");
+        private static final ParseField MAX_RETRY_DELAY_FIELD = new ParseField("max_retry_delay");
         private static final ParseField IDLE_SHARD_RETRY_DELAY = new ParseField("idle_shard_retry_delay");
         private static final ConstructingObjectParser<Request, String> PARSER = new ConstructingObjectParser<>(NAME, true,
             (args, followerIndex) -> {
@@ -75,8 +78,8 @@ public final class FollowIndexAction extends Action<AcknowledgedResponse> {
             PARSER.declareInt(ConstructingObjectParser.optionalConstructorArg(), MAX_WRITE_BUFFER_SIZE);
             PARSER.declareField(
                     ConstructingObjectParser.optionalConstructorArg(),
-                    (p, c) -> TimeValue.parseTimeValue(p.text(), MAX_RETRY_DELAY.getPreferredName()),
-                    MAX_RETRY_DELAY,
+                    (p, c) -> TimeValue.parseTimeValue(p.text(), MAX_RETRY_DELAY_FIELD.getPreferredName()),
+                MAX_RETRY_DELAY_FIELD,
                     ObjectParser.ValueType.STRING);
             PARSER.declareField(
                     ConstructingObjectParser.optionalConstructorArg(),
@@ -202,7 +205,7 @@ public final class FollowIndexAction extends Action<AcknowledgedResponse> {
                 throw new IllegalArgumentException(MAX_WRITE_BUFFER_SIZE.getPreferredName() + " must be larger than 0");
             }
 
-            final TimeValue actualRetryTimeout = maxRetryDelay == null ? DEFAULT_RETRY_TIMEOUT : maxRetryDelay;
+            final TimeValue actualRetryTimeout = maxRetryDelay == null ? DEFAULT_MAX_RETRY_DELAY : maxRetryDelay;
             final TimeValue actualIdleShardRetryDelay = idleShardRetryDelay == null ? DEFAULT_IDLE_SHARD_RETRY_DELAY : idleShardRetryDelay;
 
             this.leaderIndex = leaderIndex;
@@ -222,7 +225,20 @@ public final class FollowIndexAction extends Action<AcknowledgedResponse> {
 
         @Override
         public ActionRequestValidationException validate() {
-            return null;
+            ActionRequestValidationException validationException = null;
+
+            if (maxRetryDelay.millis() <= 0) {
+                String message = "[" + MAX_RETRY_DELAY_FIELD.getPreferredName() + "] must be positive but was [" +
+                    maxRetryDelay.getStringRep() + "]";
+                validationException = addValidationError(message, validationException);
+            }
+            if (maxRetryDelay.millis() > FollowIndexAction.MAX_RETRY_DELAY.millis()) {
+                String message = "[" + MAX_RETRY_DELAY_FIELD.getPreferredName() + "] must be less than [" + MAX_RETRY_DELAY +
+                    "] but was [" + maxRetryDelay.getStringRep() + "]";
+                validationException = addValidationError(message, validationException);
+            }
+
+            return validationException;
         }
 
         @Override
@@ -264,7 +280,7 @@ public final class FollowIndexAction extends Action<AcknowledgedResponse> {
                 builder.field(MAX_WRITE_BUFFER_SIZE.getPreferredName(), maxWriteBufferSize);
                 builder.field(MAX_CONCURRENT_READ_BATCHES.getPreferredName(), maxConcurrentReadBatches);
                 builder.field(MAX_CONCURRENT_WRITE_BATCHES.getPreferredName(), maxConcurrentWriteBatches);
-                builder.field(MAX_RETRY_DELAY.getPreferredName(), maxRetryDelay.getStringRep());
+                builder.field(MAX_RETRY_DELAY_FIELD.getPreferredName(), maxRetryDelay.getStringRep());
                 builder.field(IDLE_SHARD_RETRY_DELAY.getPreferredName(), idleShardRetryDelay.getStringRep());
             }
             builder.endObject();

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

@@ -94,10 +94,25 @@ public class PutAutoFollowPatternAction extends Action<AcknowledgedResponse> {
         public ActionRequestValidationException validate() {
             ActionRequestValidationException validationException = null;
             if (leaderClusterAlias == null) {
-                validationException = addValidationError("leaderClusterAlias is missing", validationException);
+                validationException = addValidationError("[" + LEADER_CLUSTER_ALIAS_FIELD.getPreferredName() +
+                    "] is missing", validationException);
             }
             if (leaderIndexPatterns == null || leaderIndexPatterns.isEmpty()) {
-                validationException = addValidationError("leaderIndexPatterns is missing", validationException);
+                validationException = addValidationError("[" + LEADER_INDEX_PATTERNS_FIELD.getPreferredName() +
+                    "] is missing", validationException);
+            }
+            if (maxRetryDelay != null) {
+                if (maxRetryDelay.millis() <= 0) {
+                    String message = "[" + AutoFollowPattern.MAX_RETRY_DELAY.getPreferredName() + "] must be positive but was [" +
+                        maxRetryDelay.getStringRep() + "]";
+                    validationException = addValidationError(message, validationException);
+                }
+                if (maxRetryDelay.millis() > FollowIndexAction.MAX_RETRY_DELAY.millis()) {
+                    String message = "[" + AutoFollowPattern.MAX_RETRY_DELAY.getPreferredName() + "] must be less than [" +
+                        FollowIndexAction.MAX_RETRY_DELAY +
+                        "] but was [" + maxRetryDelay.getStringRep() + "]";
+                    validationException = addValidationError(message, validationException);
+                }
             }
             return validationException;
         }

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

@@ -5,12 +5,18 @@
  */
 package org.elasticsearch.xpack.core.ccr.action;
 
+import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.test.AbstractStreamableXContentTestCase;
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Collections;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
 
 public class PutAutoFollowPatternRequestTests extends AbstractStreamableXContentTestCase<PutAutoFollowPatternAction.Request> {
 
@@ -60,4 +66,39 @@ public class PutAutoFollowPatternRequestTests extends AbstractStreamableXContent
         }
         return request;
     }
+
+    public void testValidate() {
+        PutAutoFollowPatternAction.Request request = new PutAutoFollowPatternAction.Request();
+        ActionRequestValidationException validationException = request.validate();
+        assertThat(validationException, notNullValue());
+        assertThat(validationException.getMessage(), containsString("[leader_cluster_alias] is missing"));
+
+        request.setLeaderClusterAlias("_alias");
+        validationException = request.validate();
+        assertThat(validationException, notNullValue());
+        assertThat(validationException.getMessage(), containsString("[leader_index_patterns] is missing"));
+
+        request.setLeaderIndexPatterns(Collections.emptyList());
+        validationException = request.validate();
+        assertThat(validationException, notNullValue());
+        assertThat(validationException.getMessage(), containsString("[leader_index_patterns] is missing"));
+
+        request.setLeaderIndexPatterns(Collections.singletonList("logs-*"));
+        validationException = request.validate();
+        assertThat(validationException, nullValue());
+
+        request.setMaxRetryDelay(TimeValue.ZERO);
+        validationException = request.validate();
+        assertThat(validationException, notNullValue());
+        assertThat(validationException.getMessage(), containsString("[max_retry_delay] must be positive but was [0ms]"));
+
+        request.setMaxRetryDelay(TimeValue.timeValueMinutes(10));
+        validationException = request.validate();
+        assertThat(validationException, notNullValue());
+        assertThat(validationException.getMessage(), containsString("[max_retry_delay] must be less than [5m] but was [10m]"));
+
+        request.setMaxRetryDelay(TimeValue.timeValueMinutes(1));
+        validationException = request.validate();
+        assertThat(validationException, nullValue());
+    }
 }