Disable indexing of the WatchRecord.watch_exection.input_result

This input_result may contain different types for the same field names. This will cause mapping failures when we try to write the updated watch record.
This change disables this field and adds a test in the `ManualExecutionTests` to test this case.

Fixes elastic/elasticsearch#307

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

src/main/resources/watch_history.json

@@ -38,7 +38,13 @@
         },
         "watch_execution" : {
           "type" : "object",
-          "dynamic" : true
+          "dynamic" : true,
+          "properties" : {
+            "input_result" : {
+              "type" : "object",
+              "enabled" : false
+            }
+          }
         },
         "meta" : {
           "type" : "object",

src/test/java/org/elasticsearch/watcher/execution/ManualExecutionTests.java

@@ -12,13 +12,18 @@ import org.elasticsearch.watcher.client.WatchSourceBuilder;
 import org.elasticsearch.watcher.condition.always.AlwaysCondition;
 import org.elasticsearch.watcher.history.HistoryStore;
 import org.elasticsearch.watcher.history.WatchRecord;
+import org.elasticsearch.watcher.input.simple.SimpleInput;
 import org.elasticsearch.watcher.test.AbstractWatcherIntegrationTests;
 import org.elasticsearch.watcher.transport.actions.get.GetWatchRequest;
 import org.elasticsearch.watcher.transport.actions.put.PutWatchRequest;
 import org.elasticsearch.watcher.transport.actions.put.PutWatchResponse;
+import org.elasticsearch.watcher.watch.Payload;
 import org.elasticsearch.watcher.watch.Watch;
 import org.junit.Test;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
 import static org.elasticsearch.watcher.actions.ActionBuilders.loggingAction;
 import static org.elasticsearch.watcher.client.WatchSourceBuilders.watchBuilder;
@@ -119,4 +124,40 @@ public class ManualExecutionTests extends AbstractWatcherIntegrationTests {
         }
 
     }
+
+    @Test
+    public void testDifferentAlternativeInputs() throws Exception {
+        ensureWatcherStarted();
+        WatchSourceBuilder watchBuilder = watchBuilder()
+                .trigger(schedule(cron("0 0 0 1 * ? 2099")))
+                .addAction("log", loggingAction("foobar"));
+
+        PutWatchResponse putWatchResponse = watcherClient().putWatch(new PutWatchRequest("_id", watchBuilder)).actionGet();
+        assertThat(putWatchResponse.getVersion(), greaterThan(0L));
+        refresh();
+        assertThat(watcherClient().getWatch(new GetWatchRequest("_id")).actionGet().isFound(), equalTo(true));
+
+        Map<String, Object> map1 = new HashMap<>();
+        map1.put("foo", "bar");
+
+        Map<String, Object> map2 = new HashMap<>();
+        map2.put("foo", map1);
+
+        ManualExecutionContext.Builder ctxBuilder1 = ManualExecutionContext.builder(watchService().getWatch("_id"));
+        ctxBuilder1.simulateActions("_all");
+        ctxBuilder1.withInput(new SimpleInput.Result(new Payload.Simple(map1)));
+        ctxBuilder1.recordExecution(true);
+
+        WatchRecord watchRecord1 = executionService().execute(ctxBuilder1.build());
+
+        ManualExecutionContext.Builder ctxBuilder2 = ManualExecutionContext.builder(watchService().getWatch("_id"));
+        ctxBuilder2.simulateActions("_all");
+        ctxBuilder2.withInput(new SimpleInput.Result(new Payload.Simple(map2)));
+        ctxBuilder2.recordExecution(true);
+
+        WatchRecord watchRecord2 = executionService().execute(ctxBuilder2.build());
+
+        assertThat(watchRecord1.execution().inputResult().payload().data().get("foo").toString(), equalTo("bar"));
+        assertThat(watchRecord2.execution().inputResult().payload().data().get("foo"), instanceOf(Map.class));
+    }
 }