[ML][Data Frames] unify validation exceptions between PUT/_preview (#44983) (#45012)

* [ML][Data Frames] unify validation exceptions between PUT/_preview

* addressing PR comments

x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/action/TransportPreviewDataFrameTransformAction.java

@@ -104,6 +104,19 @@ public class TransportPreviewDataFrameTransformAction extends
         }
 
         Pivot pivot = new Pivot(config.getPivotConfig());
+        try {
+            pivot.validateConfig();
+        } catch (ElasticsearchStatusException e) {
+            listener.onFailure(
+                new ElasticsearchStatusException(DataFrameMessages.REST_PUT_DATA_FRAME_FAILED_TO_VALIDATE_DATA_FRAME_CONFIGURATION,
+                    e.status(),
+                    e));
+            return;
+        } catch (Exception e) {
+            listener.onFailure(new ElasticsearchStatusException(
+                DataFrameMessages.REST_PUT_DATA_FRAME_FAILED_TO_VALIDATE_DATA_FRAME_CONFIGURATION, RestStatus.INTERNAL_SERVER_ERROR, e));
+            return;
+        }
 
         getPreview(pivot, config.getSource(), config.getDestination().getPipeline(), config.getDestination().getIndex(), listener);
     }

x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/action/TransportPutDataFrameTransformAction.java

@@ -27,6 +27,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.license.LicenseUtils;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.ClientHelper;
@@ -212,18 +213,33 @@ public class TransportPutDataFrameTransformAction extends TransportMasterNodeAct
         // <2> Put our transform
         ActionListener<Boolean> pivotValidationListener = ActionListener.wrap(
             validationResult -> dataFrameTransformsConfigManager.putTransformConfiguration(config, putTransformConfigurationListener),
-            validationException -> listener.onFailure(
+            validationException -> {
-                    new RuntimeException(DataFrameMessages.REST_PUT_DATA_FRAME_FAILED_TO_VALIDATE_DATA_FRAME_CONFIGURATION,
+                if (validationException instanceof ElasticsearchStatusException) {
-                        validationException))
+                    listener.onFailure(new ElasticsearchStatusException(
+                        DataFrameMessages.REST_PUT_DATA_FRAME_FAILED_TO_VALIDATE_DATA_FRAME_CONFIGURATION,
+                        ((ElasticsearchStatusException)validationException).status(),
+                        validationException));
+                } else {
+                    listener.onFailure(new ElasticsearchStatusException(
+                        DataFrameMessages.REST_PUT_DATA_FRAME_FAILED_TO_VALIDATE_DATA_FRAME_CONFIGURATION,
+                        RestStatus.INTERNAL_SERVER_ERROR,
+                        validationException));
+                }
+            }
         );
 
         try {
             pivot.validateConfig();
-        } catch (Exception e) {
+        } catch (ElasticsearchStatusException e) {
-            listener.onFailure(
+            listener.onFailure(new ElasticsearchStatusException(
-                new RuntimeException(DataFrameMessages.REST_PUT_DATA_FRAME_FAILED_TO_VALIDATE_DATA_FRAME_CONFIGURATION,
+                DataFrameMessages.REST_PUT_DATA_FRAME_FAILED_TO_VALIDATE_DATA_FRAME_CONFIGURATION,
+                e.status(),
                 e));
             return;
+        } catch (Exception e) {
+            listener.onFailure(new ElasticsearchStatusException(
+                DataFrameMessages.REST_PUT_DATA_FRAME_FAILED_TO_VALIDATE_DATA_FRAME_CONFIGURATION, RestStatus.INTERNAL_SERVER_ERROR, e));
+            return;
         }
 
         if (request.isDeferValidation()) {

x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/pivot/Pivot.java

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.dataframe.transforms.pivot;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.SearchAction;
 import org.elasticsearch.action.search.SearchRequest;
@@ -72,7 +73,7 @@ public class Pivot {
     public void validateConfig() {
         for (AggregationBuilder agg : config.getAggregationConfig().getAggregatorFactories()) {
             if (Aggregations.isSupportedByDataframe(agg.getType()) == false) {
-                throw new RuntimeException("Unsupported aggregation type [" + agg.getType() + "]");
+                throw new ElasticsearchStatusException("Unsupported aggregation type [" + agg.getType() + "]", RestStatus.BAD_REQUEST);
             }
         }
     }
@@ -82,15 +83,17 @@ public class Pivot {
 
         client.execute(SearchAction.INSTANCE, searchRequest, ActionListener.wrap(response -> {
             if (response == null) {
-                listener.onFailure(new RuntimeException("Unexpected null response from test query"));
+                listener.onFailure(new ElasticsearchStatusException("Unexpected null response from test query",
+                    RestStatus.SERVICE_UNAVAILABLE));
                 return;
             }
             if (response.status() != RestStatus.OK) {
-                listener.onFailure(new RuntimeException("Unexpected status from response of test query: "+ response.status()));
+                listener.onFailure(new ElasticsearchStatusException("Unexpected status from response of test query: " + response.status(),
+                    response.status()));
                 return;
             }
             listener.onResponse(true);
-        }, e -> listener.onFailure(new RuntimeException("Failed to test query", e))));
+        }, e -> listener.onFailure(new ElasticsearchStatusException("Failed to test query", RestStatus.SERVICE_UNAVAILABLE, e))));
     }
 
     public void deduceMappings(Client client, SourceConfig sourceConfig, final ActionListener<Map<String, String>> listener) {

x-pack/plugin/data-frame/src/test/java/org/elasticsearch/xpack/dataframe/transforms/pivot/PivotTests.java

@@ -7,6 +7,7 @@
 package org.elasticsearch.xpack.dataframe.transforms.pivot;
 
 import org.apache.lucene.search.TotalHits;
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequest;
@@ -144,7 +145,7 @@ public class PivotTests extends ESTestCase {
             AggregationConfig aggregationConfig = getAggregationConfig(agg);
 
             Pivot pivot = new Pivot(getValidPivotConfig(aggregationConfig));
-            RuntimeException ex = expectThrows(RuntimeException.class, pivot::validateConfig);
+            ElasticsearchException ex = expectThrows(ElasticsearchException.class, pivot::validateConfig);
             assertThat("expected aggregations to be unsupported, but they were", ex, is(notNullValue()));
         }
     }

x-pack/plugin/src/test/resources/rest-api-spec/test/data_frame/preview_transforms.yml

@@ -252,3 +252,35 @@ setup:
               }
             }
           }
+---
+"Test preview with unsupported agg":
+  - do:
+      catch: bad_request
+      data_frame.preview_data_frame_transform:
+        body: >
+          {
+            "source": { "index": "airline-data" },
+            "dest": { "pipeline": "missing-pipeline" },
+            "pivot": {
+              "group_by": {
+                "time": {"date_histogram": {"fixed_interval": "1h", "field": "time"}}},
+              "aggs": {
+                "vals": {"terms": {"field":"airline"}}
+              }
+            }
+          }
+  - do:
+      catch: /Unsupported aggregation type \[terms\]/
+      data_frame.preview_data_frame_transform:
+        body: >
+          {
+            "source": { "index": "airline-data" },
+            "dest": { "pipeline": "missing-pipeline" },
+            "pivot": {
+              "group_by": {
+                "time": {"date_histogram": {"fixed_interval": "1h", "field": "time"}}},
+              "aggs": {
+                "vals": {"terms": {"field":"airline"}}
+              }
+            }
+          }