Ensure we protect Collections obtained from scripts from self-referencing (elastic/x-pack-elasticsearch#3681)

Self referencing maps can cause SOE if they are iterated ie. in their toString methods. This chance adds some protected to the usage of those collections.
see elastic/elasticsearch#28335

Original commit: elastic/x-pack-elasticsearch@c4f1089c74

plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/execute/TransportExecuteWatchAction.java

@@ -16,6 +16,7 @@ import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.routing.Preference;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentType;
@@ -109,8 +110,14 @@ public class TransportExecuteWatchAction extends WatcherTransportAction<ExecuteW
     private void executeWatch(ExecuteWatchRequest request, ActionListener<ExecuteWatchResponse> listener,
                               Watch watch, boolean knownWatch) {
 
-        threadPool.executor(XPackField.WATCHER).submit(() -> {
-            try {
+        threadPool.executor(XPackField.WATCHER).submit(new AbstractRunnable() {
+            @Override
+            public void onFailure(Exception e) {
+                listener.onFailure(e);
+            }
+
+            @Override
+            protected void doRun() throws Exception {
                 // ensure that the headers from the incoming request are used instead those of the stored watch
                 // otherwise the watch would run as the user who stored the watch, but it needs to be run as the user who
                 // executes this request
@@ -141,8 +148,6 @@ public class TransportExecuteWatchAction extends WatcherTransportAction<ExecuteW
 
                 record.toXContent(builder, WatcherParams.builder().hideSecrets(true).debug(request.isDebug()).build());
                 listener.onResponse(new ExecuteWatchResponse(record.id().value(), builder.bytes(), XContentType.JSON));
-            } catch (IOException e) {
-                listener.onFailure(e);
             }
         });
     }

qa/smoke-test-watcher-with-painless/src/test/resources/rest-api-spec/test/watcher_painless/40_exception.yml

@@ -34,3 +34,77 @@
 
   - is_true: error.script_stack
   - match: { status: 500 }
+
+---
+"Test painless exceptions are returned when logging a broken response":
+  - do:
+      cluster.health:
+          wait_for_status: green
+
+  - do:
+      xpack.watcher.execute_watch:
+        body:  >
+          {
+            "watch" : {
+              "trigger": {
+                "schedule": {
+                  "interval": "1d"
+                }
+              },
+              "input": {
+                "simple": {
+                  "foo": "bar"
+                }
+              },
+              "actions": {
+                "my-logging": {
+                  "transform": {
+                    "script": {
+                      "source": "def x = [:] ; def y = [:] ; x.a = y ; y.a = x ; return x"
+                    }
+                  },
+                  "logging": {
+                    "text": "{{ctx}}"
+                  }
+                }
+              }
+            }
+          }
+
+  - match: { watch_record.watch_id: "_inlined_" }
+  - match: { watch_record.trigger_event.type: "manual" }
+  - match: { watch_record.state: "executed" }
+  - match: { watch_record.result.actions.0.status: "failure" }
+  - match: { watch_record.result.actions.0.error.caused_by.caused_by.type: "illegal_argument_exception" }
+  - match: { watch_record.result.actions.0.error.caused_by.caused_by.reason: "Iterable object is self-referencing itself" }
+
+  - do:
+      catch: bad_request
+      xpack.watcher.execute_watch:
+        body: >
+          {
+            "watch": {
+              "trigger": {
+                "schedule": {
+                  "interval": "10s"
+                }
+              },
+              "input": {
+                "simple": {
+                  "foo": "bar"
+                }
+              },
+              "actions": {
+                "my-logging": {
+                  "transform": {
+                    "script": {
+                      "source": "def x = [:] ; def y = [:] ; x.a = y ; y.a = x ; return x"
+                    }
+                  },
+                  "logging": {
+                    "text": "{{#join}}ctx.payload{{/join}}"
+                  }
+                }
+              }
+            }
+          }