Log more information in debug when synonyms fail updates (#102946) (#103020)

Related to: #102261

In test failures, we are not receiving any information around the bulk
indexing cause stacktrace, just the message. 

This adds debug logging and grabs the first stacktrace over all indices.
Additionally, the logger groups by the failure message in an effort to
find unique failures over all the indices.

Author: benwtrent

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/synonyms/40_synonyms_sets_get.yml

@@ -23,8 +23,21 @@ setup:
         body:
           synonyms_set:
             - synonyms: "pc, computer"
+  # set logging to debug for issue: https://github.com/elastic/elasticsearch/issues/102261
+  - do:
+      cluster.put_settings:
+        body:
+          persistent:
+            logger.org.elasticsearch.synonyms: DEBUG
 
 ---
+teardown:
+  - do:
+      cluster.put_settings:
+        body:
+          persistent:
+            logger.org.elasticsearch.synonyms: null
+---
 "List synonyms set":
   - do:
       synonyms.get_synonyms_sets: { }

server/src/main/java/org/elasticsearch/synonyms/SynonymsManagementAPIService.java

@@ -8,6 +8,8 @@
 
 package org.elasticsearch.synonyms;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.ResourceNotFoundException;
@@ -62,6 +64,8 @@
  * Manages synonyms performing operations on the system index
  */
 public class SynonymsManagementAPIService {
+
+    private static final Logger logger = LogManager.getLogger(SynonymsManagementAPIService.class);
     private static final String SYNONYMS_INDEX_NAME_PATTERN = ".synonyms-*";
     private static final int SYNONYMS_INDEX_FORMAT = 2;
     private static final String SYNONYMS_INDEX_CONCRETE_NAME = ".synonyms-" + SYNONYMS_INDEX_FORMAT;
@@ -234,11 +238,30 @@ private static SynonymRule sourceMapToSynonymRule(Map<String, Object> docSourceA
         return new SynonymRule((String) docSourceAsMap.get(SYNONYM_RULE_ID_FIELD), (String) docSourceAsMap.get(SYNONYMS_FIELD));
     }
 
+    private static void logUniqueFailureMessagesWithIndices(List<BulkItemResponse.Failure> bulkFailures) {
+        // check if logger is at least debug
+        if (logger.isDebugEnabled() == false) {
+            return;
+        }
+        Map<String, List<BulkItemResponse.Failure>> uniqueFailureMessages = bulkFailures.stream()
+            .collect(Collectors.groupingBy(BulkItemResponse.Failure::getMessage));
+        // log each unique failure with their associated indices and the first stacktrace
+        uniqueFailureMessages.forEach((failureMessage, failures) -> {
+            logger.debug(
+                "Error updating synonyms: [{}], indices: [{}], stacktrace: [{}]",
+                failureMessage,
+                failures.stream().map(BulkItemResponse.Failure::getIndex).collect(Collectors.joining(",")),
+                ExceptionsHelper.formatStackTrace(failures.get(0).getCause().getStackTrace())
+            );
+        });
+    }
+
     public void putSynonymsSet(String synonymSetId, SynonymRule[] synonymsSet, ActionListener<SynonymsReloadResult> listener) {
         deleteSynonymsSetObjects(synonymSetId, listener.delegateFailure((deleteByQueryResponseListener, bulkDeleteResponse) -> {
             boolean created = bulkDeleteResponse.getDeleted() == 0;
             final List<BulkItemResponse.Failure> bulkDeleteFailures = bulkDeleteResponse.getBulkFailures();
             if (bulkDeleteFailures.isEmpty() == false) {
+                logUniqueFailureMessagesWithIndices(bulkDeleteFailures);
                 listener.onFailure(
                     new ElasticsearchException(
                         "Error updating synonyms: "
@@ -264,6 +287,12 @@ public void putSynonymsSet(String synonymSetId, SynonymRule[] synonymsSet, Actio
             bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
                 .execute(deleteByQueryResponseListener.delegateFailure((bulkInsertResponseListener, bulkInsertResponse) -> {
                     if (bulkInsertResponse.hasFailures()) {
+                        logUniqueFailureMessagesWithIndices(
+                            Arrays.stream(bulkInsertResponse.getItems())
+                                .filter(BulkItemResponse::isFailed)
+                                .map(BulkItemResponse::getFailure)
+                                .collect(Collectors.toList())
+                        );
                         bulkInsertResponseListener.onFailure(
                             new ElasticsearchException("Error updating synonyms: " + bulkInsertResponse.buildFailureMessage())
                         );