[CCR] Do not unnecessarily wrap fetch exception in a ElasticSearch exception and (#33777)

* [CCR] Do not unnecessarily wrap fetch exception in a ElasticSearch exception and
properly map fetch_exception.exception field as object.

The extra caused by level is not necessary here:

```
"fetch_exceptions": [
              {
                "from_seq_no": 1,
                "retries": 106,
                "exception": {
                  "type": "exception",
                  "reason": "[index1] IndexNotFoundException[no such index]",
                  "caused_by": {
                    "type": "index_not_found_exception",
                    "reason": "no such index",
                    "index_uuid": "_na_",
                    "index": "index1"
                  }
                }
              }
            ],
```

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

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.ccr.action;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.support.TransportActions;
 import org.elasticsearch.common.Randomness;
 import org.elasticsearch.common.collect.Tuple;
@@ -246,7 +247,7 @@ public abstract class ShardFollowNodeTask extends AllocatedPersistentTask {
                     synchronized (ShardFollowNodeTask.this) {
                         totalFetchTimeMillis += TimeUnit.NANOSECONDS.toMillis(relativeTimeProvider.getAsLong() - startTime);
                         numberOfFailedFetches++;
-                        fetchExceptions.put(from, Tuple.tuple(retryCounter, new ElasticsearchException(e)));
+                        fetchExceptions.put(from, Tuple.tuple(retryCounter, ExceptionsHelper.convertToElastic(e)));
                     }
                     handleFailure(e, retryCounter, () -> sendShardChangesRequest(from, maxOperationCount, maxRequiredSeqNo, retryCounter));
                 });

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

@@ -198,12 +198,10 @@ public class ShardFollowNodeTaskTests extends ESTestCase {
                 final Map.Entry<Long, Tuple<Integer, ElasticsearchException>> entry = status.fetchExceptions().entrySet().iterator().next();
                 assertThat(entry.getValue().v1(), equalTo(Math.toIntExact(retryCounter.get())));
                 assertThat(entry.getKey(), equalTo(0L));
-                assertThat(entry.getValue().v2(), instanceOf(ElasticsearchException.class));
-                assertNotNull(entry.getValue().v2().getCause());
-                assertThat(entry.getValue().v2().getCause(), instanceOf(ShardNotFoundException.class));
-                final ShardNotFoundException cause = (ShardNotFoundException) entry.getValue().v2().getCause();
-                assertThat(cause.getShardId().getIndexName(), equalTo("leader_index"));
-                assertThat(cause.getShardId().getId(), equalTo(0));
+                assertThat(entry.getValue().v2(), instanceOf(ShardNotFoundException.class));
+                final ShardNotFoundException shardNotFoundException = (ShardNotFoundException) entry.getValue().v2();
+                assertThat(shardNotFoundException.getShardId().getIndexName(), equalTo("leader_index"));
+                assertThat(shardNotFoundException.getShardId().getId(), equalTo(0));
             }
             retryCounter.incrementAndGet();
         };

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/monitoring/collector/ccr/CcrStatsMonitoringDocTests.java

@@ -238,10 +238,17 @@ public class CcrStatsMonitoringDocTests extends BaseMonitoringDocTestCase<CcrSta
             } else {
                 // Manual test specific object fields and if not just fail:
                 if (fieldName.equals("fetch_exceptions")) {
+                    assertThat(fieldType, equalTo("nested"));
                     assertThat(((Map<?, ?>) fieldMapping.get("properties")).size(), equalTo(3));
                     assertThat(XContentMapValues.extractValue("properties.from_seq_no.type", fieldMapping), equalTo("long"));
                     assertThat(XContentMapValues.extractValue("properties.retries.type", fieldMapping), equalTo("integer"));
-                    assertThat(XContentMapValues.extractValue("properties.exception.type", fieldMapping), equalTo("text"));
+                    assertThat(XContentMapValues.extractValue("properties.exception.type", fieldMapping), equalTo("object"));
+
+                    Map<?, ?> exceptionFieldMapping =
+                        (Map<?, ?>) XContentMapValues.extractValue("properties.exception.properties", fieldMapping);
+                    assertThat(exceptionFieldMapping.size(), equalTo(2));
+                    assertThat(XContentMapValues.extractValue("type.type", exceptionFieldMapping), equalTo("keyword"));
+                    assertThat(XContentMapValues.extractValue("reason.type", exceptionFieldMapping), equalTo("text"));
                 } else {
                     fail("unexpected field value type [" + fieldValue.getClass() + "] for field [" + fieldName + "]");
                 }

x-pack/plugin/core/src/main/resources/monitoring-es.json

@@ -983,6 +983,7 @@
               "type": "long"
             },
             "fetch_exceptions": {
+              "type": "nested",
               "properties": {
                 "from_seq_no": {
                   "type": "long"
@@ -991,7 +992,15 @@
                   "type": "integer"
                 },
                 "exception": {
-                  "type": "text"
+                  "type": "object",
+                  "properties": {
+                    "type" : {
+                      "type": "keyword"
+                    },
+                    "reason": {
+                      "type": "text"
+                    }
+                  }
                 }
               }
             },