[Transform] Make it possible to clear retention policy on an existing transform

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/NullRetentionPolicyConfig.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.transform.transforms;
+
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xcontent.NamedXContentRegistry;
+import org.elasticsearch.xcontent.ParseField;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.core.deprecation.DeprecationIssue;
+
+import java.io.IOException;
+import java.util.function.Consumer;
+
+/**
+ * {@link NullRetentionPolicyConfig} is the implementation of {@link RetentionPolicyConfig} used when the user explicitly sets the
+ * retention_policy to {@code null} in the _update request:
+ *
+ * POST _transform/some-transform/_update
+ * {
+ *   "retention_policy": null
+ * }
+ *
+ * This is treated *differently* than simply omitting retention_policy from the request as it instructs the API to clear existing
+ * retention_policy from some-transform.
+ */
+public class NullRetentionPolicyConfig implements RetentionPolicyConfig {
+
+    public static final ParseField NAME = new ParseField("null_retention_policy");
+    public static final NullRetentionPolicyConfig INSTANCE = new NullRetentionPolicyConfig();
+
+    private NullRetentionPolicyConfig() {}
+
+    @Override
+    public ActionRequestValidationException validate(ActionRequestValidationException validationException) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void checkForDeprecations(String id, NamedXContentRegistry namedXContentRegistry, Consumer<DeprecationIssue> onDeprecation) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public String getWriteableName() {
+        return NAME.getPreferredName();
+    }
+
+    @Override
+    public void writeTo(final StreamOutput out) throws IOException {}
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfigUpdate.java

@@ -63,11 +63,15 @@ public class TransformConfigUpdate implements Writeable {
         PARSER.declareString(optionalConstructorArg(), TransformField.DESCRIPTION);
         PARSER.declareObject(optionalConstructorArg(), (p, c) -> SettingsConfig.fromXContent(p, false), TransformField.SETTINGS);
         PARSER.declareObject(optionalConstructorArg(), (p, c) -> p.mapOrdered(), TransformField.METADATA);
-        PARSER.declareNamedObject(
-            optionalConstructorArg(),
-            (p, c, n) -> p.namedObject(RetentionPolicyConfig.class, n, c),
-            TransformField.RETENTION_POLICY
-        );
+        PARSER.declareObjectOrNull(optionalConstructorArg(), (p, c) -> {
+            XContentParser.Token token = p.nextToken();
+            assert token == XContentParser.Token.FIELD_NAME;
+            String currentName = p.currentName();
+            RetentionPolicyConfig namedObject = p.namedObject(RetentionPolicyConfig.class, currentName, c);
+            token = p.nextToken();
+            assert token == XContentParser.Token.END_OBJECT;
+            return namedObject;
+        }, NullRetentionPolicyConfig.INSTANCE, TransformField.RETENTION_POLICY);
     }
 
     private final SourceConfig source;
@@ -299,7 +303,11 @@ public TransformConfig apply(TransformConfig config) {
             builder.setMetadata(metadata);
         }
         if (retentionPolicyConfig != null) {
-            builder.setRetentionPolicyConfig(retentionPolicyConfig);
+            if (NullRetentionPolicyConfig.INSTANCE.equals(retentionPolicyConfig)) {
+                builder.setRetentionPolicyConfig(null);
+            } else {
+                builder.setRetentionPolicyConfig(retentionPolicyConfig);
+            }
         }
 
         builder.setVersion(Version.CURRENT);

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/AbstractSerializingTransformTestCase.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.ToXContent.Params;
+import org.elasticsearch.xpack.core.transform.transforms.NullRetentionPolicyConfig;
 import org.elasticsearch.xpack.core.transform.transforms.RetentionPolicyConfig;
 import org.elasticsearch.xpack.core.transform.transforms.SyncConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TimeRetentionPolicyConfig;
@@ -69,6 +70,13 @@ public void registerNamedObjects() {
                 TimeRetentionPolicyConfig::new
             )
         );
+        namedWriteables.add(
+            new NamedWriteableRegistry.Entry(
+                RetentionPolicyConfig.class,
+                NullRetentionPolicyConfig.NAME.getPreferredName(),
+                in -> NullRetentionPolicyConfig.INSTANCE
+            )
+        );
 
         List<NamedXContentRegistry.Entry> namedXContents = searchModule.getNamedXContents();
         namedXContents.add(

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/action/AbstractWireSerializingTransformTestCase.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xpack.core.transform.TransformField;
 import org.elasticsearch.xpack.core.transform.TransformNamedXContentProvider;
+import org.elasticsearch.xpack.core.transform.transforms.NullRetentionPolicyConfig;
 import org.elasticsearch.xpack.core.transform.transforms.RetentionPolicyConfig;
 import org.elasticsearch.xpack.core.transform.transforms.SyncConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TimeRetentionPolicyConfig;
@@ -52,6 +53,13 @@ public void registerNamedObjects() {
                 TimeRetentionPolicyConfig::new
             )
         );
+        namedWriteables.add(
+            new NamedWriteableRegistry.Entry(
+                RetentionPolicyConfig.class,
+                NullRetentionPolicyConfig.NAME.getPreferredName(),
+                in -> NullRetentionPolicyConfig.INSTANCE
+            )
+        );
 
         List<NamedXContentRegistry.Entry> namedXContents = searchModule.getNamedXContents();
         namedXContents.addAll(new TransformNamedXContentProvider().getNamedXContentParsers());

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/transforms/TransformConfigUpdateTests.java

@@ -31,6 +31,8 @@
 import static org.elasticsearch.xpack.core.transform.transforms.TransformConfigTests.randomSyncConfig;
 import static org.elasticsearch.xpack.core.transform.transforms.TransformConfigTests.randomTransformConfig;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
 
 public class TransformConfigUpdateTests extends AbstractWireSerializingTransformTestCase<TransformConfigUpdate> {
 
@@ -43,7 +45,7 @@ public static TransformConfigUpdate randomTransformConfigUpdate() {
             randomBoolean() ? null : randomAlphaOfLengthBetween(1, 1000),
             randomBoolean() ? null : SettingsConfigTests.randomSettingsConfig(),
             randomBoolean() ? null : randomMetadata(),
-            randomBoolean() ? null : randomRetentionPolicyConfig()
+            randomBoolean() ? null : randomBoolean() ? randomRetentionPolicyConfig() : NullRetentionPolicyConfig.INSTANCE
         );
     }
 
@@ -144,6 +146,40 @@ public void testApply() {
         assertThat(updatedConfig.getVersion(), equalTo(Version.CURRENT));
     }
 
+    public void testApplyRetentionPolicy() {
+        TransformConfig config = TransformConfigTests.randomTransformConfig();
+
+        RetentionPolicyConfig timeRetentionPolicyConfig = new TimeRetentionPolicyConfig("field", TimeValue.timeValueDays(1));
+        TransformConfigUpdate setRetentionPolicy = new TransformConfigUpdate(
+            null,
+            null,
+            null,
+            null,
+            null,
+            null,
+            null,
+            timeRetentionPolicyConfig
+        );
+        config = setRetentionPolicy.apply(config);
+        assertThat(config.getRetentionPolicyConfig(), is(equalTo(timeRetentionPolicyConfig)));
+
+        TransformConfigUpdate clearRetentionPolicy = new TransformConfigUpdate(
+            null,
+            null,
+            null,
+            null,
+            null,
+            null,
+            null,
+            NullRetentionPolicyConfig.INSTANCE
+        );
+        config = clearRetentionPolicy.apply(config);
+        assertThat(config.getRetentionPolicyConfig(), is(nullValue()));
+
+        config = setRetentionPolicy.apply(config);
+        assertThat(config.getRetentionPolicyConfig(), is(equalTo(timeRetentionPolicyConfig)));
+    }
+
     public void testApplySettings() {
         TransformConfig config = new TransformConfig(
             "time-transform",
@@ -354,9 +390,13 @@ private void toXContent(TransformConfigUpdate update, XContentBuilder builder) t
             builder.field(TransformField.METADATA.getPreferredName(), update.getMetadata());
         }
         if (update.getRetentionPolicyConfig() != null) {
-            builder.startObject(TransformField.RETENTION_POLICY.getPreferredName());
-            builder.field(update.getRetentionPolicyConfig().getWriteableName(), update.getRetentionPolicyConfig());
-            builder.endObject();
+            if (NullRetentionPolicyConfig.INSTANCE.equals(update.getRetentionPolicyConfig())) {
+                builder.nullField(TransformField.RETENTION_POLICY.getPreferredName());
+            } else {
+                builder.startObject(TransformField.RETENTION_POLICY.getPreferredName());
+                builder.field(update.getRetentionPolicyConfig().getWriteableName(), update.getRetentionPolicyConfig());
+                builder.endObject();
+            }
         }
 
         builder.endObject();

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/transform/transforms_update.yml

@@ -206,6 +206,51 @@ setup:
   - match: { transforms.0.sync.time.delay: "120m" }
   - match: { transforms.0.frequency: "5s" }
 
+---
+"Test update retention policy":
+  - do:
+      transform.get_transform:
+        transform_id: "updating-airline-transform"
+  - match: { count: 1 }
+  - match: { transforms.0.id: "updating-airline-transform" }
+  - match: { transforms.0.retention_policy: null }
+
+  - do:
+      transform.update_transform:
+        transform_id: "updating-airline-transform"
+        body: >
+          {
+            "retention_policy": {
+              "time": {
+                "field": "time",
+                "max_age": "24h"
+              }
+            }
+          }
+
+  - do:
+      transform.get_transform:
+        transform_id: "updating-airline-transform"
+  - match: { count: 1 }
+  - match: { transforms.0.id: "updating-airline-transform" }
+  - match: { transforms.0.retention_policy.time.field: "time" }
+  - match: { transforms.0.retention_policy.time.max_age: "24h" }
+
+  - do:
+      transform.update_transform:
+        transform_id: "updating-airline-transform"
+        body: >
+          {
+            "retention_policy": null
+          }
+
+  - do:
+      transform.get_transform:
+        transform_id: "updating-airline-transform"
+  - match: { count: 1 }
+  - match: { transforms.0.id: "updating-airline-transform" }
+  - match: { transforms.0.retention_policy: null }
+
 ---
 "Test transform where dest is included in source":
   - do: