Globally disable AUTO_CLOSE_JSON_CONTENT. (#15880)

* Globally disable AUTO_CLOSE_JSON_CONTENT.

This JsonGenerator feature is on by default. It causes problems with code
like this:

  try (JsonGenerator jg = ...) {
    jg.writeStartArray();
    for (x : xs) {
      jg.writeObject(x);
    }
    jg.writeEndArray();
  }

If a jg.writeObject call fails due to some problem with the data it's
reading, the JsonGenerator will write the end array marker automatically
when closed as part of the try-with-resources. If the generator is writing
to a stream where the reader does not have some other mechanism to realize
that an exception was thrown, this leads the reader to believe that the
array is complete when it actually isn't.

Prior to this patch, we disabled AUTO_CLOSE_JSON_CONTENT for JSON-wrapped
SQL result formats in #11685, which fixed an issue where such results
could be erroneously interpreted as complete. This patch fixes a similar
issue with task reports, and all similar issues that may exist elsewhere,
by disabling the feature globally.

* Update test.

indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskReportSerdeTest.java

@@ -19,9 +19,12 @@
 
 package org.apache.druid.indexing.common.task;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.io.Files;
 import org.apache.druid.indexer.IngestionState;
 import org.apache.druid.indexing.common.IngestionStatsAndErrorsTaskReport;
 import org.apache.druid.indexing.common.IngestionStatsAndErrorsTaskReportData;
@@ -34,6 +37,7 @@
 import org.junit.rules.TemporaryFolder;
 
 import java.io.File;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 
 public class TaskReportSerdeTest
@@ -47,6 +51,7 @@ public TaskReportSerdeTest()
   {
     TestUtils testUtils = new TestUtils();
     jsonMapper = testUtils.getTestObjectMapper();
+    jsonMapper.registerSubtypes(ExceptionalTaskReport.class);
   }
 
   @Test
@@ -87,4 +92,47 @@ public void testSerde() throws Exception
     );
     Assert.assertEquals(reportMap1, reportMap2);
   }
+
+  @Test
+  public void testExceptionWhileWritingReport() throws Exception
+  {
+    final File reportFile = temporaryFolder.newFile();
+    final SingleFileTaskReportFileWriter writer = new SingleFileTaskReportFileWriter(reportFile);
+    writer.setObjectMapper(jsonMapper);
+    writer.write("theTask", ImmutableMap.of("report", new ExceptionalTaskReport()));
+
+    // Read the file, ensure it's incomplete and not valid JSON. This allows callers to determine the report was
+    // not complete when written.
+    Assert.assertEquals(
+        "{\"report\":{\"type\":\"exceptional\"",
+        Files.asCharSource(reportFile, StandardCharsets.UTF_8).read()
+    );
+  }
+
+  /**
+   * Task report that throws an exception while being serialized.
+   */
+  @JsonTypeName("exceptional")
+  private static class ExceptionalTaskReport implements TaskReport
+  {
+    @Override
+    @JsonProperty
+    public String getTaskId()
+    {
+      throw new UnsupportedOperationException("cannot serialize task ID");
+    }
+
+    @Override
+    public String getReportKey()
+    {
+      return "report";
+    }
+
+    @Override
+    @JsonProperty
+    public Object getPayload()
+    {
+      throw new UnsupportedOperationException("cannot serialize payload");
+    }
+  }
 }

processing/src/main/java/org/apache/druid/jackson/DefaultObjectMapper.java

@@ -20,6 +20,7 @@
 package org.apache.druid.jackson;
 
 import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.databind.DeserializationContext;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JavaType;
@@ -35,7 +36,6 @@
 import org.apache.druid.java.util.common.StringUtils;
 
 import javax.annotation.Nullable;
-
 import java.io.IOException;
 
 /**
@@ -81,6 +81,10 @@ public DefaultObjectMapper(JsonFactory factory, @Nullable String serviceName)
     configure(SerializationFeature.INDENT_OUTPUT, false);
     configure(SerializationFeature.FLUSH_AFTER_WRITE_VALUE, false);
 
+    // Disable automatic JSON termination, so readers can detect truncated responses when a JsonGenerator is
+    // closed after an exception is thrown while writing.
+    configure(JsonGenerator.Feature.AUTO_CLOSE_JSON_CONTENT, false);
+
     addHandler(new DefaultDeserializationProblemHandler(serviceName));
   }
 

sql/src/main/java/org/apache/druid/sql/http/ArrayWriter.java

@@ -43,9 +43,6 @@ public ArrayWriter(final OutputStream outputStream, final ObjectMapper jsonMappe
     this.serializers = jsonMapper.getSerializerProviderInstance();
     this.jsonGenerator = jsonMapper.getFactory().createGenerator(outputStream);
     this.outputStream = outputStream;
-
-    // Disable automatic JSON termination, so clients can detect truncated responses.
-    jsonGenerator.configure(JsonGenerator.Feature.AUTO_CLOSE_JSON_CONTENT, false);
   }
 
   @Override

sql/src/main/java/org/apache/druid/sql/http/ObjectWriter.java

@@ -46,9 +46,6 @@ public ObjectWriter(final OutputStream outputStream, final ObjectMapper jsonMapp
     this.serializers = jsonMapper.getSerializerProviderInstance();
     this.jsonGenerator = jsonMapper.getFactory().createGenerator(outputStream);
     this.outputStream = outputStream;
-
-    // Disable automatic JSON termination, so clients can detect truncated responses.
-    jsonGenerator.configure(JsonGenerator.Feature.AUTO_CLOSE_JSON_CONTENT, false);
   }
 
   @Override