Do not swallow fail to convert exceptions (#29043)

When converting the source for an indexing request to JSON, the
conversion can throw an I/O exception which we swallow and proceed with
logging to the slow log. The cause of the I/O exception is lost. This
commit changes this behavior and chooses to drop the entry from the slow
logs and instead lets an exception percolate up to the indexing
operation listener loop. Here, the exception will be caught and logged
at the warn level.

server/src/main/java/org/elasticsearch/index/IndexingSlowLog.java

@@ -33,6 +33,8 @@ import org.elasticsearch.index.shard.IndexingOperationListener;
 import org.elasticsearch.index.shard.ShardId;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Locale;
 import java.util.concurrent.TimeUnit;
 
 public final class IndexingSlowLog implements IndexingOperationListener {
@@ -194,6 +196,12 @@ public final class IndexingSlowLog implements IndexingOperationListener {
                 sb.append(", source[").append(Strings.cleanTruncate(source, maxSourceCharsToLog)).append("]");
             } catch (IOException e) {
                 sb.append(", source[_failed_to_convert_[").append(e.getMessage()).append("]]");
+                /*
+                 * We choose to fail to write to the slow log and instead let this percolate up to the post index listener loop where this
+                 * will be logged at the warn level.
+                 */
+                final String message = String.format(Locale.ROOT, "failed to convert source for slow log entry [%s]", sb.toString());
+                throw new UncheckedIOException(message, e);
             }
             return sb.toString();
         }

server/src/test/java/org/elasticsearch/index/IndexingSlowLogTests.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.index;
 
+import com.fasterxml.jackson.core.JsonParseException;
 import org.apache.lucene.document.NumericDocValuesField;
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
@@ -34,6 +35,7 @@ import org.elasticsearch.index.mapper.SeqNoFieldMapper;
 import org.elasticsearch.test.ESTestCase;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.hasToString;
@@ -70,9 +72,15 @@ public class IndexingSlowLogTests extends ESTestCase {
             "test", null, null, source, XContentType.JSON, null);
         p = new SlowLogParsedDocumentPrinter(index, pd, 10, true, 3);
 
-        assertThat(p.toString(), containsString("_failed_to_convert_[Unrecognized token 'invalid':"
+        final UncheckedIOException e = expectThrows(UncheckedIOException.class, p::toString);
+        assertThat(e, hasToString(containsString("_failed_to_convert_[Unrecognized token 'invalid':"
             + " was expecting ('true', 'false' or 'null')\n"
-            + " at [Source: org.elasticsearch.common.bytes.BytesReference$MarkSupportingStreamInputWrapper"));
+            + " at [Source: org.elasticsearch.common.bytes.BytesReference$MarkSupportingStreamInputWrapper")));
+        assertNotNull(e.getCause());
+        assertThat(e.getCause(), instanceOf(JsonParseException.class));
+        assertThat(e.getCause(), hasToString(containsString("Unrecognized token 'invalid':"
+                + " was expecting ('true', 'false' or 'null')\n"
+                + " at [Source: org.elasticsearch.common.bytes.BytesReference$MarkSupportingStreamInputWrapper")));
     }
 
     public void testReformatSetting() {